Add pulp-manage-db check for running workers.

pulp-manage-db command now waits to check if workers have been
stopped. If there are running workers the migration is stopped

closes #2186
https://pulp.plan.io/issues/2186

common/pulp/common/constants.py

@@ -66,6 +66,11 @@
 # The maximum number of seconds that will ever elapse before the scheduler looks for
 # new or changed schedules
 CELERY_MAX_INTERVAL = 90
+# The amount of time the migration script will wait to confirm that no processes are running.
+# This is the 90s CELERY_TICK_DEFAULT_WAIT_TIME used in Pulp version < 2.12 and a 2s buffer.
+# This ensures that the process check feature works correctly even in cases where a user
+# forgot to restart pulp_celerybeat while upgrading from Pulp 2.11 or earlier.
+MIGRATION_WAIT_TIME = 92
 
 # resource manager constants
 

docs/user-guide/release-notes/2.12.x.rst

@@ -9,6 +9,10 @@ New Features
 ------------
 
  * Task profiling can now be enabled. This will use cProfiling on an individual task and write the profile to a directory for a given task. While this can impact performance, this can enable users to get some insight into what a task is doing or use the output to give to a developer for debugging.
+ * ``pulp-manage-db`` will not continue if pulp_celerybeat, pulp_resource_manager, or pulp_workers
+   processes are running. This will prevent the user from corrupting their Pulp installation by
+   migrating with running workers. This works in both standalone and clustered installations. Pulp
+   may wait up to 92 seconds to determine if workers are running.
 
 Deprecation
 -----------

server/pulp/server/db/manage.py

@@ -1,23 +1,28 @@
 """
 This module's main() function becomes the pulp-manage-db.py script.
 """
+from datetime import datetime, timedelta
 from gettext import gettext as _
 from optparse import OptionParser
 import logging
 import os
 import sys
+import time
 import traceback
 
+from pulp.common import constants
 from pulp.plugins.loader.api import load_content_types
 from pulp.plugins.loader.manager import PluginManager
 from pulp.server import logs
 from pulp.server.db import connection
 from pulp.server.db.migrate import models
 from pulp.server.db import model
 from pulp.server.db.migrations.lib import managers
-from pulp.server.managers import factory
+from pulp.server.db.fields import UTCDateTimeField
+from pulp.server.managers import factory, status
 from pulp.server.managers.auth.role.cud import RoleManager, SUPER_USER_ROLE
 
+from pymongo.errors import ServerSelectionTimeoutError
 
 os.environ['DJANGO_SETTINGS_MODULE'] = 'pulp.server.webservices.settings'
 
@@ -191,6 +196,34 @@ def main():
         options = parse_args()
         _start_logging()
         connection.initialize(max_timeout=1)
+
+        active_workers = status.get_workers()
+        if active_workers:
+            last_worker_time = max([worker['last_heartbeat'] for worker in active_workers])
+            time_from_last_worker = UTCDateTimeField().to_python(datetime.now()) - last_worker_time
+            wait_time = timedelta(seconds=constants.MIGRATION_WAIT_TIME) - time_from_last_worker
+
+            if wait_time > timedelta(0):
+                print _('\nThe following processes might still be running:')
+                for worker in active_workers:
+                    print _('\t%s' % worker['name'])
+
+                for i in range(wait_time.seconds, 0, -1):
+                    print _('\rPlease wait %s seconds while Pulp confirms this.' % i),
+                    sys.stdout.flush()
+                    time.sleep(1)
+
+                still_active_workers = [worker for worker in status.get_workers() if
+                                        worker['last_heartbeat'] > last_worker_time]
+
+                if still_active_workers:
+                    print >> sys.stderr, _('\n\nThe following processes are still running, please'
+                                           ' stop the running workers before retrying the'
+                                           ' pulp-manage-db command.')
+                    for worker in still_active_workers:
+                        print _('\t%s' % worker['name'])
+
+                    return os.EX_SOFTWARE
         return _auto_manage_db(options)
     except UnperformedMigrationException:
         return 1
@@ -200,6 +233,10 @@ def main():
         return os.EX_DATAERR
     except models.MigrationRemovedError:
         return os.EX_SOFTWARE
+    except ServerSelectionTimeoutError:
+        _logger.info(_('Cannot connect to the database, please validate that the database is online'
+                       ' and accessible.'))
+        return os.EX_SOFTWARE
     except Exception, e:
         _logger.critical(str(e))
         _logger.critical(''.join(traceback.format_exception(*sys.exc_info())))