Ensure Celery worker gets restarted
Occasionally, Monit will get interrupted during a restart. This
pidfile check retries the update_worker call if the PID did not
change.
@@ -83,14 +83,20 @@ class ProjectPushWorkerManager(object):
self._write_worker_configuration(queue_config)
status = yield self._wait_for_stable_state()
+ pid_location = os.path.join(PID_DIR, 'celery-{}.pid'.format(self.project_id))
+ try:
+ with open(pid_location) as pidfile:
+ old_pid = int(pidfile.read().strip())
+ except IOError:
+ old_pid = None
+
# Start the worker if it doesn't exist. Restart it if it does.
if status == MonitStates.MISSING:
command = self.celery_command()
env_vars = {'APP_ID': self.project_id, 'HOST': options.load_balancers[0],
'C_FORCE_ROOT': True}
- pidfile = os.path.join(PID_DIR, 'celery-{}.pid'.format(self.project_id))
- create_config_file(self.monit_watch, command, pidfile, env_vars=env_vars,
- max_memory=CELERY_SAFE_MEMORY)
+ create_config_file(self.monit_watch, command, pid_location,
+ env_vars=env_vars, max_memory=CELERY_SAFE_MEMORY)
logger.info('Starting push worker for {}'.format(self.project_id))
yield self.monit_operator.reload()
else:
@@ -101,6 +107,22 @@ class ProjectPushWorkerManager(object):
yield gen.with_timeout(timedelta(seconds=60), start_future,
IOLoop.current())
+ # Wait for the worker to write its new PID.
+ yield gen.sleep(1)
+ try:
+ with open(pid_location) as pidfile:
+ new_pid = int(pidfile.read().strip())
+ except IOError:
+ new_pid = None
+
+ # Occasionally, Monit will get interrupted during a restart. Retry the
+ # restart if the Celery worker PID is the same.
+ if new_pid == old_pid:
+ logger.warning(
+ '{} worker PID did not change. Restarting it.'.format(self.project_id))
+ yield gen.sleep(1)
+ yield self.update_worker(queue_config)
+
@gen.coroutine
def stop_worker(self):
""" Removes the monit configuration for the project's push worker. """
