Investigate if we can replace gunicornmontor with uvicorn.run() #43035

It is most likely that we no longer need gunicornmontor or UvicornMonitor anymore. @ashb 's suggestion is for Airflow uvicorn.run() should be enough.

Whoever takes this GitHub issue should verify the same and replace it if not needed.

The code:

• airflow/airflow/cli/commands/fastapi_api_command.py

  Lines 159 to 190 in f38d56d

  	def monitor_gunicorn(gunicorn_master_proc: psutil.Process | subprocess.Popen) -> NoReturn:
  	# Register signal handlers
  	signal.signal(signal.SIGINT, lambda signum, _: kill_proc(signum, gunicorn_master_proc))
  	signal.signal(signal.SIGTERM, lambda signum, _: kill_proc(signum, gunicorn_master_proc))
  	# These run forever until SIG{INT, TERM, KILL, ...} signal is sent
  	GunicornMonitor(
  	gunicorn_master_pid=gunicorn_master_proc.pid,
  	num_workers_expected=num_workers,
  	master_timeout=120,
  	worker_refresh_interval=30,
  	worker_refresh_batch_size=1,
  	reload_on_plugin_change=False,
  	).start()
  	def start_and_monitor_gunicorn(args):
  	if args.daemon:
  	subprocess.Popen(run_args, close_fds=True)
  	# Reading pid of gunicorn master as it will be different that
  	# the one of process spawned above.
  	gunicorn_master_proc_pid = None
  	while not gunicorn_master_proc_pid:
  	sleep(0.1)
  	gunicorn_master_proc_pid = read_pid_from_pidfile(pid_file)
  	# Run Gunicorn monitor
  	gunicorn_master_proc = psutil.Process(gunicorn_master_proc_pid)
  	monitor_gunicorn(gunicorn_master_proc)
  	else:
  	with subprocess.Popen(run_args, close_fds=True) as gunicorn_master_proc:
  	monitor_gunicorn(gunicorn_master_proc)

• airflow/airflow/cli/commands/webserver_command.py

  Lines 49 to 107 in f38d56d

  	class GunicornMonitor(LoggingMixin):
  	"""
  	Runs forever.
  	Monitoring the child processes of @gunicorn_master_proc and restarting
  	workers occasionally or when files in the plug-in directory has been modified.
  	Each iteration of the loop traverses one edge of this state transition
  	diagram, where each state (node) represents
  	[ num_ready_workers_running / num_workers_running ]. We expect most time to
  	be spent in [n / n]. `bs` is the setting webserver.worker_refresh_batch_size.
  	The horizontal transition at ? happens after the new worker parses all the
  	dags (so it could take a while!)
  	V ────────────────────────────────────────────────────────────────────────┐
  	[n / n] ──TTIN──> [ [n, n+bs) / n + bs ] ────?───> [n + bs / n + bs] ──TTOU─┘
  	^ ^───────────────┘
  	│
  	│ ┌────────────────v
  	└──────┴────── [ [0, n) / n ] <─── start
  	We change the number of workers by sending TTIN and TTOU to the gunicorn
  	master process, which increases and decreases the number of child workers
  	respectively. Gunicorn guarantees that on TTOU workers are terminated
  	gracefully and that the oldest worker is terminated.
  	:param gunicorn_master_pid: PID for the main Gunicorn process
  	:param num_workers_expected: Number of workers to run the Gunicorn web server
  	:param master_timeout: Number of seconds the webserver waits before killing gunicorn master that
  	doesn't respond
  	:param worker_refresh_interval: Number of seconds to wait before refreshing a batch of workers.
  	:param worker_refresh_batch_size: Number of workers to refresh at a time. When set to 0, worker
  	refresh is disabled. When nonzero, airflow periodically refreshes webserver workers by
  	bringing up new ones and killing old ones.
  	:param reload_on_plugin_change: If set to True, Airflow will track files in plugins_folder directory.
  	When it detects changes, then reload the gunicorn.
  	"""
  	def __init__(
  	self,
  	gunicorn_master_pid: int,
  	num_workers_expected: int,
  	master_timeout: int,
  	worker_refresh_interval: int,
  	worker_refresh_batch_size: int,
  	reload_on_plugin_change: bool,
  	):
  	super().__init__()
  	self.gunicorn_master_proc = psutil.Process(gunicorn_master_pid)
  	self.num_workers_expected = num_workers_expected
  	self.master_timeout = master_timeout
  	self.worker_refresh_interval = worker_refresh_interval
  	self.worker_refresh_batch_size = worker_refresh_batch_size
  	self.reload_on_plugin_change = reload_on_plugin_change
  	self._num_workers_running = 0
  	self._num_ready_workers_running = 0
  	self._last_refresh_time = time.monotonic() if worker_refresh_interval > 0 else None
  	self._last_plugin_state = self._generate_plugin_state() if reload_on_plugin_change else None
  	self._restart_on_next_plugin_check = False