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 2020/05/24 18:51:37 UTC
[GitHub] [airflow] mik-laj opened a new pull request #8997: Reload gunicorn when plugins has beeen changed
mik-laj opened a new pull request #8997:
URL: https://github.com/apache/airflow/pull/8997
User expect that when a new version of the plugin file is available, it will be used. This behavior is similar to DAG files. We always use latest version of DAG file. This change will also cause us to always use the latest plug-in files.
Now webserver monitor will track changes in the plugin directory and when it detects changes in this directory, it will reload the gunicorn configuration.
I didn't use inotify because not all filesystems sends valid events.
---
Make sure to mark the boxes below before creating PR: [x]
- [X] Description above provides context of the change
- [X] Unit tests coverage for changes (not needed for documentation changes)
- [X] Target Github ISSUE in description if exists
- [X] Commits follow "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)"
- [X] Relevant documentation is updated including usage instructions.
- [X] I will engage committers as explained in [Contribution Workflow Example](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#contribution-workflow-example).
---
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
Read the [Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines) for more information.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #8997: Reload gunicorn when plugins has beeen changed
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #8997:
URL: https://github.com/apache/airflow/pull/8997#discussion_r434638254
##########
File path: airflow/cli/commands/webserver_command.py
##########
@@ -83,92 +62,238 @@ def restart_workers(gunicorn_master_proc, num_workers_expected, master_timeout):
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.
- """
- def wait_until_true(fn, timeout=0):
+ :param gunicorn_master_proc: handle 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_follder directory.
+ When it detects changes, then reload the gunicorn.
+ """
+ def __init__(
+ self,
+ gunicorn_master_proc: psutil.Process,
+ num_workers_expected: int,
+ master_timeout: int,
+ worker_refresh_interval: int,
+ worker_refresh_batch_size: int,
+ reload_on_plugin_change: bool
+ ):
+ super(GunicornMonitor, self).__init__()
+ self.gunicorn_master_proc = gunicorn_master_proc
+ 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.time() 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
+
+ def _generate_plugin_state(self) -> Dict[str, float]:
+ """
+ Generate dict of filenames and last modification time of all files in settings.PLUGINS_FOLDER
+ directory.
+ """
+ if not settings.PLUGINS_FOLDER:
+ return {}
+ all_filenames: List[str] = []
+ for (root, _, filenames) in os.walk(settings.PLUGINS_FOLDER):
+ # filenames: List[str] = cast(List[str], filenames)
+ all_filenames.extend(os.path.join(root, f) for f in filenames)
Review comment:
I think it will be much better to calculate hash of the files ? I think mtime is a bad thing to use for multiple reasons - one of them is that in a number of situations you can get mtime modified when the file is not updated various sync solutions might do it). I always prefer to calculate hashes in those kind of checks.
##########
File path: airflow/cli/commands/webserver_command.py
##########
@@ -226,13 +351,13 @@ def webserver(args):
run_args = [
'gunicorn',
- '-w', str(num_workers),
- '-k', str(args.workerclass),
- '-t', str(worker_timeout),
- '-b', args.hostname + ':' + str(args.port),
- '-n', 'airflow-webserver',
- '-p', pid_file,
- '-c', 'python:airflow.www.gunicorn_config',
+ '--workers', str(num_workers),
Review comment:
:heart:
##########
File path: airflow/cli/commands/webserver_command.py
##########
@@ -83,92 +62,238 @@ def restart_workers(gunicorn_master_proc, num_workers_expected, master_timeout):
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.
- """
- def wait_until_true(fn, timeout=0):
+ :param gunicorn_master_proc: handle 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_follder directory.
+ When it detects changes, then reload the gunicorn.
+ """
+ def __init__(
+ self,
+ gunicorn_master_proc: psutil.Process,
+ num_workers_expected: int,
+ master_timeout: int,
+ worker_refresh_interval: int,
+ worker_refresh_batch_size: int,
+ reload_on_plugin_change: bool
+ ):
+ super(GunicornMonitor, self).__init__()
+ self.gunicorn_master_proc = gunicorn_master_proc
+ 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.time() 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
+
+ def _generate_plugin_state(self) -> Dict[str, float]:
+ """
+ Generate dict of filenames and last modification time of all files in settings.PLUGINS_FOLDER
+ directory.
+ """
+ if not settings.PLUGINS_FOLDER:
+ return {}
+ all_filenames: List[str] = []
+ for (root, _, filenames) in os.walk(settings.PLUGINS_FOLDER):
+ # filenames: List[str] = cast(List[str], filenames)
+ all_filenames.extend(os.path.join(root, f) for f in filenames)
+ plugin_state = {f: os.path.getmtime(f) for f in sorted(all_filenames)}
+ return plugin_state
+
+ def _get_num_ready_workers_running(self) -> int:
+ """Returns number of ready Gunicorn workers by looking for READY_PREFIX in process name"""
+ workers = psutil.Process(self.gunicorn_master_proc.pid).children()
+
+ def ready_prefix_on_cmdline(proc):
+ try:
+ cmdline = proc.cmdline()
+ if len(cmdline) > 0: # pylint: disable=len-as-condition
+ return settings.GUNICORN_WORKER_READY_PREFIX in cmdline[0]
+ except psutil.NoSuchProcess:
+ pass
+ return False
+
+ ready_workers = [proc for proc in workers if ready_prefix_on_cmdline(proc)]
+ return len(ready_workers)
+
+ def _get_num_workers_running(self) -> int:
+ """Returns number of running Gunicorn workers processes"""
+ workers = psutil.Process(self.gunicorn_master_proc.pid).children()
+ return len(workers)
+
+ def _wait_until_true(self, fn, timeout: int = 0) -> None:
"""
Sleeps until fn is true
"""
start_time = time.time()
while not fn():
if 0 < timeout <= time.time() - start_time:
raise AirflowWebServerTimeout(
- "No response from gunicorn master within {0} seconds"
- .format(timeout))
- time.sleep(0.1)
+ "No response from gunicorn master within {0} seconds".format(timeout)
+ )
+ sleep(0.1)
- def start_refresh(gunicorn_master_proc):
- batch_size = conf.getint('webserver', 'worker_refresh_batch_size')
- log.debug('%s doing a refresh of %s workers', state, batch_size)
- sys.stdout.flush()
- sys.stderr.flush()
+ def _spawn_new_workers(self, count: int) -> None:
+ """
+ Send signal to kill the worker.
+ :param count: The number of workers to spawn
+ """
excess = 0
- for _ in range(batch_size):
- gunicorn_master_proc.send_signal(signal.SIGTTIN)
+ for _ in range(count):
+ # TTIN: Increment the number of processes by one
+ self.gunicorn_master_proc.send_signal(signal.SIGTTIN)
excess += 1
- wait_until_true(lambda: num_workers_expected + excess ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
-
- try: # pylint: disable=too-many-nested-blocks
- wait_until_true(lambda: num_workers_expected ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
- while True:
- num_workers_running = get_num_workers_running(gunicorn_master_proc)
- num_ready_workers_running = \
- get_num_ready_workers_running(gunicorn_master_proc)
-
- state = '[{0} / {1}]'.format(num_ready_workers_running, num_workers_running)
-
- # Whenever some workers are not ready, wait until all workers are ready
- if num_ready_workers_running < num_workers_running:
- log.debug('%s some workers are starting up, waiting...', state)
- sys.stdout.flush()
- time.sleep(1)
-
- # Kill a worker gracefully by asking gunicorn to reduce number of workers
- elif num_workers_running > num_workers_expected:
- excess = num_workers_running - num_workers_expected
- log.debug('%s killing %s workers', state, excess)
-
- for _ in range(excess):
- gunicorn_master_proc.send_signal(signal.SIGTTOU)
- excess -= 1
- wait_until_true(lambda: num_workers_expected + excess ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
-
- # Start a new worker by asking gunicorn to increase number of workers
- elif num_workers_running == num_workers_expected:
- refresh_interval = conf.getint('webserver', 'worker_refresh_interval')
- log.debug(
- '%s sleeping for %ss starting doing a refresh...',
- state, refresh_interval
+ self._wait_until_true(
+ lambda: self.num_workers_expected + excess == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+
+ def _kill_old_workers(self, count: int) -> None:
+ """
+ Send signal to kill the worker.
+
+ :param count: The number of workers to kill
+ """
+ for _ in range(count):
+ count -= 1
+ # TTOU: Decrement the number of processes by one
+ self.gunicorn_master_proc.send_signal(signal.SIGTTOU)
+ self._wait_until_true(
+ lambda: self.num_workers_expected + count == self._get_num_workers_running(),
+ timeout=self.master_timeout)
+
+ def _reload_gunicorn(self) -> None:
+ """"
+ Send signal to reload the gunciron configuration. When gunciorn receive signals, it reload the
+ configuration, start the new worker processes with a new configuration and gracefully
+ shutdown older workers.
+ """
+ # HUP: Reload the configuration.
+ self.gunicorn_master_proc.send_signal(signal.SIGHUP)
+ sleep(1)
+ self._wait_until_true(
+ lambda: self.num_workers_expected == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+
+ def start(self) -> NoReturn:
+ """
+ Starts monitoring the webserver.
+ """
+ try: # pylint: disable=too-many-nested-blocks
+ self._wait_until_true(
+ lambda: self.num_workers_expected == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+ while True:
+ while self.gunicorn_master_proc.poll() is not None:
+ sys.exit(self.gunicorn_master_proc.returncode)
+ self._check_workers()
+ # Throttle loop
+ sleep(1)
+
+ except (AirflowWebServerTimeout, OSError) as err:
+ self.log.error(err)
+ self.log.error("Shutting down webserver")
+ try:
+ self.gunicorn_master_proc.terminate()
+ self.gunicorn_master_proc.wait()
+ finally:
+ sys.exit(1)
+
+ def _check_workers(self) -> None:
+ num_workers_running = self._get_num_workers_running()
+ num_ready_workers_running = self._get_num_ready_workers_running()
+
+ # Whenever some workers are not ready, wait until all workers are ready
+ if num_ready_workers_running < num_workers_running:
+ self.log.debug(
+ '[%d / %d] Some workers are starting up, waiting...',
+ num_ready_workers_running, num_workers_running
+ )
+ sleep(1)
+ return
+
+ # If there are too many workers, then kill a worker gracefully by asking gunicorn to reduce
+ # number of workers
+ if num_workers_running > self.num_workers_expected:
+ excess = min(num_workers_running - self.num_workers_expected, self.worker_refresh_batch_size)
+ self.log.debug(
+ '[%d / %d] Killing %s workers', num_ready_workers_running, num_workers_running, excess
+ )
+ self._kill_old_workers(excess)
+ return
+
+ # If there are too few workers, start a new worker by asking gunicorn
+ # to increase number of workers
+ if num_workers_running < self.num_workers_expected:
+ self.log.error(
+ "[%d / %d] Some workers seem to have died and gunicorn did not restart "
+ "them as expected",
+ num_ready_workers_running, num_workers_running
+ )
+ sleep(10)
+ num_workers_running = self._get_num_workers_running()
+ if num_workers_running < self.num_workers_expected:
+ new_worker_count = min(
+ num_workers_running - self.worker_refresh_batch_size, self.worker_refresh_batch_size
)
- time.sleep(refresh_interval)
- start_refresh(gunicorn_master_proc)
-
- else:
- # num_ready_workers_running == num_workers_running < num_workers_expected
- log.error((
- "%s some workers seem to have died and gunicorn"
- "did not restart them as expected"
- ), state)
- time.sleep(10)
- if len(
- psutil.Process(gunicorn_master_proc.pid).children()
- ) < num_workers_expected:
- start_refresh(gunicorn_master_proc)
- except (AirflowWebServerTimeout, OSError) as err:
- log.error(err)
- log.error("Shutting down webserver")
- try:
- gunicorn_master_proc.terminate()
- gunicorn_master_proc.wait()
- finally:
- sys.exit(1)
+ self.log.debug(
+ '[%d / %d] Spawning %d workers',
+ num_ready_workers_running, num_workers_running, new_worker_count
+ )
+ self._spawn_new_workers(num_workers_running)
+ return
+
+ # Now the number of running and expected worker should be equal
+
+ # If workers should be restarted periodically.
+ if self.worker_refresh_interval > 0 and self._last_refresh_time:
+ # and we refreshed the workers a long time ago, refresh the workers
+ last_refresh_diff = (time.time() - self._last_refresh_time)
+ if self.worker_refresh_interval < last_refresh_diff:
+ num_new_workers = self.worker_refresh_batch_size
+ self.log.debug(
+ '[%d / %d] Starting doing a refresh. Starting %d workers.',
+ num_ready_workers_running, num_workers_running, num_new_workers
+ )
+ self._spawn_new_workers(num_new_workers)
+ self._last_refresh_time = time.time()
+ return
+
+ # if we should check the directory with the plugin,
+ if self.reload_on_plugin_change:
+ # compare the previous and current contents of the directory
+ new_state = self._generate_plugin_state()
+ # If changed, wait until its content is fully saved.
+ if new_state != self._last_plugin_state:
Review comment:
I think we should check if _last_plugin_state was actually set before and not restart everything then otherwise we will pretty much always restart everything immediately after starting. That will make the startup time longer for webserver.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #8997: Reload gunicorn when plugins has beeen changed
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8997:
URL: https://github.com/apache/airflow/pull/8997#discussion_r446969807
##########
File path: airflow/cli/commands/webserver_command.py
##########
@@ -83,92 +63,247 @@ def restart_workers(gunicorn_master_proc, num_workers_expected, master_timeout):
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.
- """
- def wait_until_true(fn, timeout=0):
+ :param gunicorn_master_proc: handle 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_follder directory.
+ When it detects changes, then reload the gunicorn.
+ """
+ def __init__(
+ self,
+ gunicorn_master_proc: psutil.Process,
+ num_workers_expected: int,
+ master_timeout: int,
+ worker_refresh_interval: int,
+ worker_refresh_batch_size: int,
+ reload_on_plugin_change: bool
+ ):
+ super(GunicornMonitor, self).__init__()
+ self.gunicorn_master_proc = gunicorn_master_proc
+ 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.time() 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
+
+ def _generate_plugin_state(self) -> Dict[str, float]:
+ """
+ Generate dict of filenames and last modification time of all files in settings.PLUGINS_FOLDER
+ directory.
+ """
+ if not settings.PLUGINS_FOLDER:
+ return {}
+
+ all_filenames: List[str] = []
+ for (root, _, filenames) in os.walk(settings.PLUGINS_FOLDER):
+ all_filenames.extend(os.path.join(root, f) for f in filenames)
+ plugin_state = {f: self._get_file_hash(f) for f in sorted(all_filenames)}
+ return plugin_state
+
+ @staticmethod
+ def _get_file_hash(fname: str):
+ """Calculate MD5 hash for file"""
+ hash_md5 = hashlib.md5()
+ with open(fname, "rb") as f:
+ for chunk in iter(lambda: f.read(4096), b""):
+ hash_md5.update(chunk)
+ return hash_md5.hexdigest()
+
+ def _get_num_ready_workers_running(self) -> int:
+ """Returns number of ready Gunicorn workers by looking for READY_PREFIX in process name"""
+ workers = psutil.Process(self.gunicorn_master_proc.pid).children()
+
+ def ready_prefix_on_cmdline(proc):
+ try:
+ cmdline = proc.cmdline()
+ if len(cmdline) > 0: # pylint: disable=len-as-condition
+ return settings.GUNICORN_WORKER_READY_PREFIX in cmdline[0]
+ except psutil.NoSuchProcess:
+ pass
+ return False
+
+ ready_workers = [proc for proc in workers if ready_prefix_on_cmdline(proc)]
+ return len(ready_workers)
+
+ def _get_num_workers_running(self) -> int:
+ """Returns number of running Gunicorn workers processes"""
+ workers = psutil.Process(self.gunicorn_master_proc.pid).children()
+ return len(workers)
+
+ def _wait_until_true(self, fn, timeout: int = 0) -> None:
"""
Sleeps until fn is true
"""
start_time = time.time()
while not fn():
if 0 < timeout <= time.time() - start_time:
raise AirflowWebServerTimeout(
- "No response from gunicorn master within {0} seconds"
- .format(timeout))
- time.sleep(0.1)
+ "No response from gunicorn master within {0} seconds".format(timeout)
+ )
+ sleep(0.1)
- def start_refresh(gunicorn_master_proc):
- batch_size = conf.getint('webserver', 'worker_refresh_batch_size')
- log.debug('%s doing a refresh of %s workers', state, batch_size)
- sys.stdout.flush()
- sys.stderr.flush()
+ def _spawn_new_workers(self, count: int) -> None:
+ """
+ Send signal to kill the worker.
+ :param count: The number of workers to spawn
+ """
excess = 0
- for _ in range(batch_size):
- gunicorn_master_proc.send_signal(signal.SIGTTIN)
+ for _ in range(count):
+ # TTIN: Increment the number of processes by one
+ self.gunicorn_master_proc.send_signal(signal.SIGTTIN)
excess += 1
- wait_until_true(lambda: num_workers_expected + excess ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
-
- try: # pylint: disable=too-many-nested-blocks
- wait_until_true(lambda: num_workers_expected ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
- while True:
- num_workers_running = get_num_workers_running(gunicorn_master_proc)
- num_ready_workers_running = \
- get_num_ready_workers_running(gunicorn_master_proc)
-
- state = '[{0} / {1}]'.format(num_ready_workers_running, num_workers_running)
-
- # Whenever some workers are not ready, wait until all workers are ready
- if num_ready_workers_running < num_workers_running:
- log.debug('%s some workers are starting up, waiting...', state)
- sys.stdout.flush()
- time.sleep(1)
-
- # Kill a worker gracefully by asking gunicorn to reduce number of workers
- elif num_workers_running > num_workers_expected:
- excess = num_workers_running - num_workers_expected
- log.debug('%s killing %s workers', state, excess)
-
- for _ in range(excess):
- gunicorn_master_proc.send_signal(signal.SIGTTOU)
- excess -= 1
- wait_until_true(lambda: num_workers_expected + excess ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
-
- # Start a new worker by asking gunicorn to increase number of workers
- elif num_workers_running == num_workers_expected:
- refresh_interval = conf.getint('webserver', 'worker_refresh_interval')
- log.debug(
- '%s sleeping for %ss starting doing a refresh...',
- state, refresh_interval
+ self._wait_until_true(
+ lambda: self.num_workers_expected + excess == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+
+ def _kill_old_workers(self, count: int) -> None:
+ """
+ Send signal to kill the worker.
+
+ :param count: The number of workers to kill
+ """
+ for _ in range(count):
+ count -= 1
+ # TTOU: Decrement the number of processes by one
+ self.gunicorn_master_proc.send_signal(signal.SIGTTOU)
+ self._wait_until_true(
+ lambda: self.num_workers_expected + count == self._get_num_workers_running(),
+ timeout=self.master_timeout)
+
+ def _reload_gunicorn(self) -> None:
+ """
+ Send signal to reload the gunciron configuration. When gunciorn receive signals, it reload the
+ configuration, start the new worker processes with a new configuration and gracefully
+ shutdown older workers.
+ """
+ # HUP: Reload the configuration.
+ self.gunicorn_master_proc.send_signal(signal.SIGHUP)
+ sleep(1)
+ self._wait_until_true(
+ lambda: self.num_workers_expected == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+
+ def start(self) -> NoReturn:
+ """
+ Starts monitoring the webserver.
+ """
+ try: # pylint: disable=too-many-nested-blocks
+ self._wait_until_true(
+ lambda: self.num_workers_expected == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+ while True:
+ while self.gunicorn_master_proc.poll() is not None:
+ sys.exit(self.gunicorn_master_proc.returncode)
+ self._check_workers()
+ # Throttle loop
+ sleep(1)
+
+ except (AirflowWebServerTimeout, OSError) as err:
+ self.log.error(err)
+ self.log.error("Shutting down webserver")
+ try:
+ self.gunicorn_master_proc.terminate()
+ self.gunicorn_master_proc.wait()
+ finally:
+ sys.exit(1)
+
+ def _check_workers(self) -> None:
+ num_workers_running = self._get_num_workers_running()
+ num_ready_workers_running = self._get_num_ready_workers_running()
+
+ # Whenever some workers are not ready, wait until all workers are ready
+ if num_ready_workers_running < num_workers_running:
+ self.log.debug(
+ '[%d / %d] Some workers are starting up, waiting...',
+ num_ready_workers_running, num_workers_running
+ )
+ sleep(1)
+ return
+
+ # If there are too many workers, then kill a worker gracefully by asking gunicorn to reduce
+ # number of workers
+ if num_workers_running > self.num_workers_expected:
+ excess = min(num_workers_running - self.num_workers_expected, self.worker_refresh_batch_size)
+ self.log.debug(
+ '[%d / %d] Killing %s workers', num_ready_workers_running, num_workers_running, excess
+ )
+ self._kill_old_workers(excess)
+ return
+
+ # If there are too few workers, start a new worker by asking gunicorn
+ # to increase number of workers
+ if num_workers_running < self.num_workers_expected:
+ self.log.error(
+ "[%d / %d] Some workers seem to have died and gunicorn did not restart "
+ "them as expected",
+ num_ready_workers_running, num_workers_running
+ )
+ sleep(10)
+ num_workers_running = self._get_num_workers_running()
+ if num_workers_running < self.num_workers_expected:
+ new_worker_count = min(
+ num_workers_running - self.worker_refresh_batch_size, self.worker_refresh_batch_size
)
- time.sleep(refresh_interval)
- start_refresh(gunicorn_master_proc)
-
- else:
- # num_ready_workers_running == num_workers_running < num_workers_expected
- log.error((
- "%s some workers seem to have died and gunicorn"
- "did not restart them as expected"
- ), state)
- time.sleep(10)
- if len(
- psutil.Process(gunicorn_master_proc.pid).children()
- ) < num_workers_expected:
- start_refresh(gunicorn_master_proc)
- except (AirflowWebServerTimeout, OSError) as err:
- log.error(err)
- log.error("Shutting down webserver")
- try:
- gunicorn_master_proc.terminate()
- gunicorn_master_proc.wait()
- finally:
- sys.exit(1)
+ self.log.debug(
+ '[%d / %d] Spawning %d workers',
+ num_ready_workers_running, num_workers_running, new_worker_count
+ )
+ self._spawn_new_workers(num_workers_running)
+ return
+
+ # Now the number of running and expected worker should be equal
+
+ # If workers should be restarted periodically.
+ if self.worker_refresh_interval > 0 and self._last_refresh_time:
+ # and we refreshed the workers a long time ago, refresh the workers
+ last_refresh_diff = (time.time() - self._last_refresh_time)
+ if self.worker_refresh_interval < last_refresh_diff:
+ num_new_workers = self.worker_refresh_batch_size
+ self.log.debug(
+ '[%d / %d] Starting doing a refresh. Starting %d workers.',
+ num_ready_workers_running, num_workers_running, num_new_workers
+ )
+ self._spawn_new_workers(num_new_workers)
+ self._last_refresh_time = time.time()
+ return
+
+ # if we should check the directory with the plugin,
+ if self.reload_on_plugin_change:
+ # compare the previous and current contents of the directory
+ new_state = self._generate_plugin_state()
+ # If changed, wait until its content is fully saved.
+ if new_state != self._last_plugin_state:
+ self.log.debug(
+ '[%d / %d] Plugins folder changed. The gunicorn will be restarted the next time the '
+ 'plugin directory is checked, if there is no change in it.',
+ num_ready_workers_running, num_workers_running
+ )
+ self._restart_on_next_plugin_check = True
+ self._last_plugin_state = new_state
+ elif self._restart_on_next_plugin_check:
Review comment:
Should we change the if clause order?
```python
if self._restart_on_next_plugin_check:
...
elif new_state != self._last_plugin_state:
...
```
If the plugins will be changing between each loop then we will never really restart workers.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #8997: Reload gunicorn when plugins has beeen changed
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #8997:
URL: https://github.com/apache/airflow/pull/8997#discussion_r446968698
##########
File path: airflow/cli/commands/webserver_command.py
##########
@@ -83,92 +63,247 @@ def restart_workers(gunicorn_master_proc, num_workers_expected, master_timeout):
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.
- """
- def wait_until_true(fn, timeout=0):
+ :param gunicorn_master_proc: handle 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_follder directory.
+ When it detects changes, then reload the gunicorn.
+ """
+ def __init__(
+ self,
+ gunicorn_master_proc: psutil.Process,
+ num_workers_expected: int,
+ master_timeout: int,
+ worker_refresh_interval: int,
+ worker_refresh_batch_size: int,
+ reload_on_plugin_change: bool
+ ):
+ super(GunicornMonitor, self).__init__()
+ self.gunicorn_master_proc = gunicorn_master_proc
+ 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.time() 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
+
+ def _generate_plugin_state(self) -> Dict[str, float]:
+ """
+ Generate dict of filenames and last modification time of all files in settings.PLUGINS_FOLDER
+ directory.
+ """
+ if not settings.PLUGINS_FOLDER:
+ return {}
+
+ all_filenames: List[str] = []
+ for (root, _, filenames) in os.walk(settings.PLUGINS_FOLDER):
+ all_filenames.extend(os.path.join(root, f) for f in filenames)
+ plugin_state = {f: self._get_file_hash(f) for f in sorted(all_filenames)}
Review comment:
In the next steps we compare two dictionaries, so yes. I am not sure if os.walk returns items in a sorted manner. I am not sure if os.walk returns items in a sorted way in each case. The documentation does not describe this.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #8997: Reload gunicorn when plugins has beeen changed
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #8997:
URL: https://github.com/apache/airflow/pull/8997#discussion_r447063973
##########
File path: airflow/cli/commands/webserver_command.py
##########
@@ -83,92 +63,247 @@ def restart_workers(gunicorn_master_proc, num_workers_expected, master_timeout):
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.
- """
- def wait_until_true(fn, timeout=0):
+ :param gunicorn_master_proc: handle 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_follder directory.
+ When it detects changes, then reload the gunicorn.
+ """
+ def __init__(
+ self,
+ gunicorn_master_proc: psutil.Process,
+ num_workers_expected: int,
+ master_timeout: int,
+ worker_refresh_interval: int,
+ worker_refresh_batch_size: int,
+ reload_on_plugin_change: bool
+ ):
+ super(GunicornMonitor, self).__init__()
+ self.gunicorn_master_proc = gunicorn_master_proc
+ 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.time() 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
+
+ def _generate_plugin_state(self) -> Dict[str, float]:
+ """
+ Generate dict of filenames and last modification time of all files in settings.PLUGINS_FOLDER
+ directory.
+ """
+ if not settings.PLUGINS_FOLDER:
+ return {}
+
+ all_filenames: List[str] = []
+ for (root, _, filenames) in os.walk(settings.PLUGINS_FOLDER):
+ all_filenames.extend(os.path.join(root, f) for f in filenames)
+ plugin_state = {f: self._get_file_hash(f) for f in sorted(all_filenames)}
Review comment:
It's not sorted. It returns entries in the sequence stored in the filesystem directory index, so it's usually in order of creation of the directories but when you delete a dir/recreate another they might reuse empty slot so for all purposes it's essentially random.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #8997: Reload gunicorn when plugins has beeen changed
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #8997:
URL: https://github.com/apache/airflow/pull/8997#discussion_r447063973
##########
File path: airflow/cli/commands/webserver_command.py
##########
@@ -83,92 +63,247 @@ def restart_workers(gunicorn_master_proc, num_workers_expected, master_timeout):
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.
- """
- def wait_until_true(fn, timeout=0):
+ :param gunicorn_master_proc: handle 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_follder directory.
+ When it detects changes, then reload the gunicorn.
+ """
+ def __init__(
+ self,
+ gunicorn_master_proc: psutil.Process,
+ num_workers_expected: int,
+ master_timeout: int,
+ worker_refresh_interval: int,
+ worker_refresh_batch_size: int,
+ reload_on_plugin_change: bool
+ ):
+ super(GunicornMonitor, self).__init__()
+ self.gunicorn_master_proc = gunicorn_master_proc
+ 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.time() 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
+
+ def _generate_plugin_state(self) -> Dict[str, float]:
+ """
+ Generate dict of filenames and last modification time of all files in settings.PLUGINS_FOLDER
+ directory.
+ """
+ if not settings.PLUGINS_FOLDER:
+ return {}
+
+ all_filenames: List[str] = []
+ for (root, _, filenames) in os.walk(settings.PLUGINS_FOLDER):
+ all_filenames.extend(os.path.join(root, f) for f in filenames)
+ plugin_state = {f: self._get_file_hash(f) for f in sorted(all_filenames)}
Review comment:
It's not sorted. It returns entries in the sequence stored in the filesystem directory index, so it's usually in order of creation of the directories but when you delete a dir/recreate another they might reuse missing slot so for all purposes it's essentially random.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj merged pull request #8997: Reload gunicorn when plugins has beeen changed
Posted by GitBox <gi...@apache.org>.
mik-laj merged pull request #8997:
URL: https://github.com/apache/airflow/pull/8997
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #8997: Reload gunicorn when plugins has beeen changed
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #8997:
URL: https://github.com/apache/airflow/pull/8997#discussion_r446901937
##########
File path: airflow/cli/commands/webserver_command.py
##########
@@ -83,92 +62,238 @@ def restart_workers(gunicorn_master_proc, num_workers_expected, master_timeout):
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.
- """
- def wait_until_true(fn, timeout=0):
+ :param gunicorn_master_proc: handle 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_follder directory.
+ When it detects changes, then reload the gunicorn.
+ """
+ def __init__(
+ self,
+ gunicorn_master_proc: psutil.Process,
+ num_workers_expected: int,
+ master_timeout: int,
+ worker_refresh_interval: int,
+ worker_refresh_batch_size: int,
+ reload_on_plugin_change: bool
+ ):
+ super(GunicornMonitor, self).__init__()
+ self.gunicorn_master_proc = gunicorn_master_proc
+ 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.time() 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
+
+ def _generate_plugin_state(self) -> Dict[str, float]:
+ """
+ Generate dict of filenames and last modification time of all files in settings.PLUGINS_FOLDER
+ directory.
+ """
+ if not settings.PLUGINS_FOLDER:
+ return {}
+ all_filenames: List[str] = []
+ for (root, _, filenames) in os.walk(settings.PLUGINS_FOLDER):
+ # filenames: List[str] = cast(List[str], filenames)
+ all_filenames.extend(os.path.join(root, f) for f in filenames)
+ plugin_state = {f: os.path.getmtime(f) for f in sorted(all_filenames)}
+ return plugin_state
+
+ def _get_num_ready_workers_running(self) -> int:
+ """Returns number of ready Gunicorn workers by looking for READY_PREFIX in process name"""
+ workers = psutil.Process(self.gunicorn_master_proc.pid).children()
+
+ def ready_prefix_on_cmdline(proc):
+ try:
+ cmdline = proc.cmdline()
+ if len(cmdline) > 0: # pylint: disable=len-as-condition
+ return settings.GUNICORN_WORKER_READY_PREFIX in cmdline[0]
+ except psutil.NoSuchProcess:
+ pass
+ return False
+
+ ready_workers = [proc for proc in workers if ready_prefix_on_cmdline(proc)]
+ return len(ready_workers)
+
+ def _get_num_workers_running(self) -> int:
+ """Returns number of running Gunicorn workers processes"""
+ workers = psutil.Process(self.gunicorn_master_proc.pid).children()
+ return len(workers)
+
+ def _wait_until_true(self, fn, timeout: int = 0) -> None:
"""
Sleeps until fn is true
"""
start_time = time.time()
while not fn():
if 0 < timeout <= time.time() - start_time:
raise AirflowWebServerTimeout(
- "No response from gunicorn master within {0} seconds"
- .format(timeout))
- time.sleep(0.1)
+ "No response from gunicorn master within {0} seconds".format(timeout)
+ )
+ sleep(0.1)
- def start_refresh(gunicorn_master_proc):
- batch_size = conf.getint('webserver', 'worker_refresh_batch_size')
- log.debug('%s doing a refresh of %s workers', state, batch_size)
- sys.stdout.flush()
- sys.stderr.flush()
+ def _spawn_new_workers(self, count: int) -> None:
+ """
+ Send signal to kill the worker.
+ :param count: The number of workers to spawn
+ """
excess = 0
- for _ in range(batch_size):
- gunicorn_master_proc.send_signal(signal.SIGTTIN)
+ for _ in range(count):
+ # TTIN: Increment the number of processes by one
+ self.gunicorn_master_proc.send_signal(signal.SIGTTIN)
excess += 1
- wait_until_true(lambda: num_workers_expected + excess ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
-
- try: # pylint: disable=too-many-nested-blocks
- wait_until_true(lambda: num_workers_expected ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
- while True:
- num_workers_running = get_num_workers_running(gunicorn_master_proc)
- num_ready_workers_running = \
- get_num_ready_workers_running(gunicorn_master_proc)
-
- state = '[{0} / {1}]'.format(num_ready_workers_running, num_workers_running)
-
- # Whenever some workers are not ready, wait until all workers are ready
- if num_ready_workers_running < num_workers_running:
- log.debug('%s some workers are starting up, waiting...', state)
- sys.stdout.flush()
- time.sleep(1)
-
- # Kill a worker gracefully by asking gunicorn to reduce number of workers
- elif num_workers_running > num_workers_expected:
- excess = num_workers_running - num_workers_expected
- log.debug('%s killing %s workers', state, excess)
-
- for _ in range(excess):
- gunicorn_master_proc.send_signal(signal.SIGTTOU)
- excess -= 1
- wait_until_true(lambda: num_workers_expected + excess ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
-
- # Start a new worker by asking gunicorn to increase number of workers
- elif num_workers_running == num_workers_expected:
- refresh_interval = conf.getint('webserver', 'worker_refresh_interval')
- log.debug(
- '%s sleeping for %ss starting doing a refresh...',
- state, refresh_interval
+ self._wait_until_true(
+ lambda: self.num_workers_expected + excess == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+
+ def _kill_old_workers(self, count: int) -> None:
+ """
+ Send signal to kill the worker.
+
+ :param count: The number of workers to kill
+ """
+ for _ in range(count):
+ count -= 1
+ # TTOU: Decrement the number of processes by one
+ self.gunicorn_master_proc.send_signal(signal.SIGTTOU)
+ self._wait_until_true(
+ lambda: self.num_workers_expected + count == self._get_num_workers_running(),
+ timeout=self.master_timeout)
+
+ def _reload_gunicorn(self) -> None:
+ """"
+ Send signal to reload the gunciron configuration. When gunciorn receive signals, it reload the
+ configuration, start the new worker processes with a new configuration and gracefully
+ shutdown older workers.
+ """
+ # HUP: Reload the configuration.
+ self.gunicorn_master_proc.send_signal(signal.SIGHUP)
+ sleep(1)
+ self._wait_until_true(
+ lambda: self.num_workers_expected == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+
+ def start(self) -> NoReturn:
+ """
+ Starts monitoring the webserver.
+ """
+ try: # pylint: disable=too-many-nested-blocks
+ self._wait_until_true(
+ lambda: self.num_workers_expected == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+ while True:
+ while self.gunicorn_master_proc.poll() is not None:
+ sys.exit(self.gunicorn_master_proc.returncode)
+ self._check_workers()
+ # Throttle loop
+ sleep(1)
+
+ except (AirflowWebServerTimeout, OSError) as err:
+ self.log.error(err)
+ self.log.error("Shutting down webserver")
+ try:
+ self.gunicorn_master_proc.terminate()
+ self.gunicorn_master_proc.wait()
+ finally:
+ sys.exit(1)
+
+ def _check_workers(self) -> None:
+ num_workers_running = self._get_num_workers_running()
+ num_ready_workers_running = self._get_num_ready_workers_running()
+
+ # Whenever some workers are not ready, wait until all workers are ready
+ if num_ready_workers_running < num_workers_running:
+ self.log.debug(
+ '[%d / %d] Some workers are starting up, waiting...',
+ num_ready_workers_running, num_workers_running
+ )
+ sleep(1)
+ return
+
+ # If there are too many workers, then kill a worker gracefully by asking gunicorn to reduce
+ # number of workers
+ if num_workers_running > self.num_workers_expected:
+ excess = min(num_workers_running - self.num_workers_expected, self.worker_refresh_batch_size)
+ self.log.debug(
+ '[%d / %d] Killing %s workers', num_ready_workers_running, num_workers_running, excess
+ )
+ self._kill_old_workers(excess)
+ return
+
+ # If there are too few workers, start a new worker by asking gunicorn
+ # to increase number of workers
+ if num_workers_running < self.num_workers_expected:
+ self.log.error(
+ "[%d / %d] Some workers seem to have died and gunicorn did not restart "
+ "them as expected",
+ num_ready_workers_running, num_workers_running
+ )
+ sleep(10)
+ num_workers_running = self._get_num_workers_running()
+ if num_workers_running < self.num_workers_expected:
+ new_worker_count = min(
+ num_workers_running - self.worker_refresh_batch_size, self.worker_refresh_batch_size
)
- time.sleep(refresh_interval)
- start_refresh(gunicorn_master_proc)
-
- else:
- # num_ready_workers_running == num_workers_running < num_workers_expected
- log.error((
- "%s some workers seem to have died and gunicorn"
- "did not restart them as expected"
- ), state)
- time.sleep(10)
- if len(
- psutil.Process(gunicorn_master_proc.pid).children()
- ) < num_workers_expected:
- start_refresh(gunicorn_master_proc)
- except (AirflowWebServerTimeout, OSError) as err:
- log.error(err)
- log.error("Shutting down webserver")
- try:
- gunicorn_master_proc.terminate()
- gunicorn_master_proc.wait()
- finally:
- sys.exit(1)
+ self.log.debug(
+ '[%d / %d] Spawning %d workers',
+ num_ready_workers_running, num_workers_running, new_worker_count
+ )
+ self._spawn_new_workers(num_workers_running)
+ return
+
+ # Now the number of running and expected worker should be equal
+
+ # If workers should be restarted periodically.
+ if self.worker_refresh_interval > 0 and self._last_refresh_time:
+ # and we refreshed the workers a long time ago, refresh the workers
+ last_refresh_diff = (time.time() - self._last_refresh_time)
+ if self.worker_refresh_interval < last_refresh_diff:
+ num_new_workers = self.worker_refresh_batch_size
+ self.log.debug(
+ '[%d / %d] Starting doing a refresh. Starting %d workers.',
+ num_ready_workers_running, num_workers_running, num_new_workers
+ )
+ self._spawn_new_workers(num_new_workers)
+ self._last_refresh_time = time.time()
+ return
+
+ # if we should check the directory with the plugin,
+ if self.reload_on_plugin_change:
+ # compare the previous and current contents of the directory
+ new_state = self._generate_plugin_state()
+ # If changed, wait until its content is fully saved.
+ if new_state != self._last_plugin_state:
Review comment:
The value for field _last_plugin_state is always set in the constructor when observing changes in the plugin directory is enabled.
https://github.com/apache/airflow/pull/8997/files#diff-71e5f1bb6f81ad283bca26aad652ae12R97
```
self._last_plugin_state = self._generate_plugin_state() if reload_on_plugin_change else None
```
It will not affect performance either, because it is run in a separate process. We first run gunicorn as a separate process, and then we begin to observe workers. Once all workers are up, we start to observe the plugins directory and regularly restart workers. In the meantime, when the workers start, we calculate the initial state of the directory.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #8997: Reload gunicorn when plugins has beeen changed
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8997:
URL: https://github.com/apache/airflow/pull/8997#discussion_r446956807
##########
File path: airflow/cli/commands/webserver_command.py
##########
@@ -83,92 +63,247 @@ def restart_workers(gunicorn_master_proc, num_workers_expected, master_timeout):
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.
- """
- def wait_until_true(fn, timeout=0):
+ :param gunicorn_master_proc: handle 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_follder directory.
+ When it detects changes, then reload the gunicorn.
+ """
+ def __init__(
+ self,
+ gunicorn_master_proc: psutil.Process,
+ num_workers_expected: int,
+ master_timeout: int,
+ worker_refresh_interval: int,
+ worker_refresh_batch_size: int,
+ reload_on_plugin_change: bool
+ ):
+ super(GunicornMonitor, self).__init__()
Review comment:
```suggestion
super().__init__()
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #8997: Reload gunicorn when plugins has beeen changed
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8997:
URL: https://github.com/apache/airflow/pull/8997#discussion_r446961217
##########
File path: airflow/cli/commands/webserver_command.py
##########
@@ -83,92 +63,247 @@ def restart_workers(gunicorn_master_proc, num_workers_expected, master_timeout):
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.
- """
- def wait_until_true(fn, timeout=0):
+ :param gunicorn_master_proc: handle 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_follder directory.
+ When it detects changes, then reload the gunicorn.
+ """
+ def __init__(
+ self,
+ gunicorn_master_proc: psutil.Process,
+ num_workers_expected: int,
+ master_timeout: int,
+ worker_refresh_interval: int,
+ worker_refresh_batch_size: int,
+ reload_on_plugin_change: bool
+ ):
+ super(GunicornMonitor, self).__init__()
+ self.gunicorn_master_proc = gunicorn_master_proc
+ 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.time() 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
+
+ def _generate_plugin_state(self) -> Dict[str, float]:
+ """
+ Generate dict of filenames and last modification time of all files in settings.PLUGINS_FOLDER
+ directory.
+ """
+ if not settings.PLUGINS_FOLDER:
+ return {}
+
+ all_filenames: List[str] = []
+ for (root, _, filenames) in os.walk(settings.PLUGINS_FOLDER):
+ all_filenames.extend(os.path.join(root, f) for f in filenames)
+ plugin_state = {f: self._get_file_hash(f) for f in sorted(all_filenames)}
Review comment:
```suggestion
plugin_state = {}
for (root, _, filenames) in os.walk(settings.PLUGINS_FOLDER):
all_filenames.extend(os.path.join(root, f) for f in filenames)
plugin_state[f] = self._get_file_hash(f)
```
Is this sorted necessary?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #8997: Reload gunicorn when plugins has beeen changed
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #8997:
URL: https://github.com/apache/airflow/pull/8997#discussion_r446974700
##########
File path: airflow/cli/commands/webserver_command.py
##########
@@ -83,92 +63,247 @@ def restart_workers(gunicorn_master_proc, num_workers_expected, master_timeout):
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.
- """
- def wait_until_true(fn, timeout=0):
+ :param gunicorn_master_proc: handle 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_follder directory.
+ When it detects changes, then reload the gunicorn.
+ """
+ def __init__(
+ self,
+ gunicorn_master_proc: psutil.Process,
+ num_workers_expected: int,
+ master_timeout: int,
+ worker_refresh_interval: int,
+ worker_refresh_batch_size: int,
+ reload_on_plugin_change: bool
+ ):
+ super(GunicornMonitor, self).__init__()
+ self.gunicorn_master_proc = gunicorn_master_proc
+ 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.time() 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
+
+ def _generate_plugin_state(self) -> Dict[str, float]:
+ """
+ Generate dict of filenames and last modification time of all files in settings.PLUGINS_FOLDER
+ directory.
+ """
+ if not settings.PLUGINS_FOLDER:
+ return {}
+
+ all_filenames: List[str] = []
+ for (root, _, filenames) in os.walk(settings.PLUGINS_FOLDER):
+ all_filenames.extend(os.path.join(root, f) for f in filenames)
+ plugin_state = {f: self._get_file_hash(f) for f in sorted(all_filenames)}
+ return plugin_state
+
+ @staticmethod
+ def _get_file_hash(fname: str):
+ """Calculate MD5 hash for file"""
+ hash_md5 = hashlib.md5()
+ with open(fname, "rb") as f:
+ for chunk in iter(lambda: f.read(4096), b""):
+ hash_md5.update(chunk)
+ return hash_md5.hexdigest()
+
+ def _get_num_ready_workers_running(self) -> int:
+ """Returns number of ready Gunicorn workers by looking for READY_PREFIX in process name"""
+ workers = psutil.Process(self.gunicorn_master_proc.pid).children()
+
+ def ready_prefix_on_cmdline(proc):
+ try:
+ cmdline = proc.cmdline()
+ if len(cmdline) > 0: # pylint: disable=len-as-condition
+ return settings.GUNICORN_WORKER_READY_PREFIX in cmdline[0]
+ except psutil.NoSuchProcess:
+ pass
+ return False
+
+ ready_workers = [proc for proc in workers if ready_prefix_on_cmdline(proc)]
+ return len(ready_workers)
+
+ def _get_num_workers_running(self) -> int:
+ """Returns number of running Gunicorn workers processes"""
+ workers = psutil.Process(self.gunicorn_master_proc.pid).children()
+ return len(workers)
+
+ def _wait_until_true(self, fn, timeout: int = 0) -> None:
"""
Sleeps until fn is true
"""
start_time = time.time()
while not fn():
if 0 < timeout <= time.time() - start_time:
raise AirflowWebServerTimeout(
- "No response from gunicorn master within {0} seconds"
- .format(timeout))
- time.sleep(0.1)
+ "No response from gunicorn master within {0} seconds".format(timeout)
+ )
+ sleep(0.1)
- def start_refresh(gunicorn_master_proc):
- batch_size = conf.getint('webserver', 'worker_refresh_batch_size')
- log.debug('%s doing a refresh of %s workers', state, batch_size)
- sys.stdout.flush()
- sys.stderr.flush()
+ def _spawn_new_workers(self, count: int) -> None:
+ """
+ Send signal to kill the worker.
+ :param count: The number of workers to spawn
+ """
excess = 0
- for _ in range(batch_size):
- gunicorn_master_proc.send_signal(signal.SIGTTIN)
+ for _ in range(count):
+ # TTIN: Increment the number of processes by one
+ self.gunicorn_master_proc.send_signal(signal.SIGTTIN)
excess += 1
- wait_until_true(lambda: num_workers_expected + excess ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
-
- try: # pylint: disable=too-many-nested-blocks
- wait_until_true(lambda: num_workers_expected ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
- while True:
- num_workers_running = get_num_workers_running(gunicorn_master_proc)
- num_ready_workers_running = \
- get_num_ready_workers_running(gunicorn_master_proc)
-
- state = '[{0} / {1}]'.format(num_ready_workers_running, num_workers_running)
-
- # Whenever some workers are not ready, wait until all workers are ready
- if num_ready_workers_running < num_workers_running:
- log.debug('%s some workers are starting up, waiting...', state)
- sys.stdout.flush()
- time.sleep(1)
-
- # Kill a worker gracefully by asking gunicorn to reduce number of workers
- elif num_workers_running > num_workers_expected:
- excess = num_workers_running - num_workers_expected
- log.debug('%s killing %s workers', state, excess)
-
- for _ in range(excess):
- gunicorn_master_proc.send_signal(signal.SIGTTOU)
- excess -= 1
- wait_until_true(lambda: num_workers_expected + excess ==
- get_num_workers_running(gunicorn_master_proc),
- master_timeout)
-
- # Start a new worker by asking gunicorn to increase number of workers
- elif num_workers_running == num_workers_expected:
- refresh_interval = conf.getint('webserver', 'worker_refresh_interval')
- log.debug(
- '%s sleeping for %ss starting doing a refresh...',
- state, refresh_interval
+ self._wait_until_true(
+ lambda: self.num_workers_expected + excess == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+
+ def _kill_old_workers(self, count: int) -> None:
+ """
+ Send signal to kill the worker.
+
+ :param count: The number of workers to kill
+ """
+ for _ in range(count):
+ count -= 1
+ # TTOU: Decrement the number of processes by one
+ self.gunicorn_master_proc.send_signal(signal.SIGTTOU)
+ self._wait_until_true(
+ lambda: self.num_workers_expected + count == self._get_num_workers_running(),
+ timeout=self.master_timeout)
+
+ def _reload_gunicorn(self) -> None:
+ """
+ Send signal to reload the gunciron configuration. When gunciorn receive signals, it reload the
+ configuration, start the new worker processes with a new configuration and gracefully
+ shutdown older workers.
+ """
+ # HUP: Reload the configuration.
+ self.gunicorn_master_proc.send_signal(signal.SIGHUP)
+ sleep(1)
+ self._wait_until_true(
+ lambda: self.num_workers_expected == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+
+ def start(self) -> NoReturn:
+ """
+ Starts monitoring the webserver.
+ """
+ try: # pylint: disable=too-many-nested-blocks
+ self._wait_until_true(
+ lambda: self.num_workers_expected == self._get_num_workers_running(),
+ timeout=self.master_timeout
+ )
+ while True:
+ while self.gunicorn_master_proc.poll() is not None:
+ sys.exit(self.gunicorn_master_proc.returncode)
+ self._check_workers()
+ # Throttle loop
+ sleep(1)
+
+ except (AirflowWebServerTimeout, OSError) as err:
+ self.log.error(err)
+ self.log.error("Shutting down webserver")
+ try:
+ self.gunicorn_master_proc.terminate()
+ self.gunicorn_master_proc.wait()
+ finally:
+ sys.exit(1)
+
+ def _check_workers(self) -> None:
+ num_workers_running = self._get_num_workers_running()
+ num_ready_workers_running = self._get_num_ready_workers_running()
+
+ # Whenever some workers are not ready, wait until all workers are ready
+ if num_ready_workers_running < num_workers_running:
+ self.log.debug(
+ '[%d / %d] Some workers are starting up, waiting...',
+ num_ready_workers_running, num_workers_running
+ )
+ sleep(1)
+ return
+
+ # If there are too many workers, then kill a worker gracefully by asking gunicorn to reduce
+ # number of workers
+ if num_workers_running > self.num_workers_expected:
+ excess = min(num_workers_running - self.num_workers_expected, self.worker_refresh_batch_size)
+ self.log.debug(
+ '[%d / %d] Killing %s workers', num_ready_workers_running, num_workers_running, excess
+ )
+ self._kill_old_workers(excess)
+ return
+
+ # If there are too few workers, start a new worker by asking gunicorn
+ # to increase number of workers
+ if num_workers_running < self.num_workers_expected:
+ self.log.error(
+ "[%d / %d] Some workers seem to have died and gunicorn did not restart "
+ "them as expected",
+ num_ready_workers_running, num_workers_running
+ )
+ sleep(10)
+ num_workers_running = self._get_num_workers_running()
+ if num_workers_running < self.num_workers_expected:
+ new_worker_count = min(
+ num_workers_running - self.worker_refresh_batch_size, self.worker_refresh_batch_size
)
- time.sleep(refresh_interval)
- start_refresh(gunicorn_master_proc)
-
- else:
- # num_ready_workers_running == num_workers_running < num_workers_expected
- log.error((
- "%s some workers seem to have died and gunicorn"
- "did not restart them as expected"
- ), state)
- time.sleep(10)
- if len(
- psutil.Process(gunicorn_master_proc.pid).children()
- ) < num_workers_expected:
- start_refresh(gunicorn_master_proc)
- except (AirflowWebServerTimeout, OSError) as err:
- log.error(err)
- log.error("Shutting down webserver")
- try:
- gunicorn_master_proc.terminate()
- gunicorn_master_proc.wait()
- finally:
- sys.exit(1)
+ self.log.debug(
+ '[%d / %d] Spawning %d workers',
+ num_ready_workers_running, num_workers_running, new_worker_count
+ )
+ self._spawn_new_workers(num_workers_running)
+ return
+
+ # Now the number of running and expected worker should be equal
+
+ # If workers should be restarted periodically.
+ if self.worker_refresh_interval > 0 and self._last_refresh_time:
+ # and we refreshed the workers a long time ago, refresh the workers
+ last_refresh_diff = (time.time() - self._last_refresh_time)
+ if self.worker_refresh_interval < last_refresh_diff:
+ num_new_workers = self.worker_refresh_batch_size
+ self.log.debug(
+ '[%d / %d] Starting doing a refresh. Starting %d workers.',
+ num_ready_workers_running, num_workers_running, num_new_workers
+ )
+ self._spawn_new_workers(num_new_workers)
+ self._last_refresh_time = time.time()
+ return
+
+ # if we should check the directory with the plugin,
+ if self.reload_on_plugin_change:
+ # compare the previous and current contents of the directory
+ new_state = self._generate_plugin_state()
+ # If changed, wait until its content is fully saved.
+ if new_state != self._last_plugin_state:
+ self.log.debug(
+ '[%d / %d] Plugins folder changed. The gunicorn will be restarted the next time the '
+ 'plugin directory is checked, if there is no change in it.',
+ num_ready_workers_running, num_workers_running
+ )
+ self._restart_on_next_plugin_check = True
+ self._last_plugin_state = new_state
+ elif self._restart_on_next_plugin_check:
Review comment:
This is expected behavior. Then the workers will probably be restarted according to the previous rules. After all, it is unlikely that the plugins will be modified more often than 1 per second. This condition is intended to prevent the server from restarting while the synchronization process is in progress. This is a more likely problem.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org