Make namespace deletion asynchronous

Instead of deleting a namespace synchronously as before, we now mark the namespace for deletion, disable it, and rename it. A worker then comes along and deletes the namespace in the background. This results in a *significantly* better user experience, as the namespace deletion operation now "completes" in under a second, where before it could take 10s of minutes at the worse.

Fixes https://jira.coreos.com/browse/QUAY-838

data/model/user.py

@@ -14,7 +14,7 @@ from data.database import (User, LoginService, FederatedLogin, RepositoryPermiss
                            EmailConfirmation, Role, db_for_update, random_string_generator,
                            UserRegion, ImageStorageLocation,
                            ServiceKeyApproval, OAuthApplication, RepositoryBuildTrigger,
-                           UserPromptKind, UserPrompt, UserPromptTypes)
+                           UserPromptKind, UserPrompt, UserPromptTypes, DeletedNamespace)
 from data.model import (DataModelException, InvalidPasswordException, InvalidRobotException,
                         InvalidUsernameException, InvalidEmailAddressException,
                         TooManyLoginAttemptsException, db_transaction,
@@ -835,7 +835,14 @@ def get_solely_admined_organizations(user_obj):
   return solely_admined
 
 
-def delete_user(user, queues, force=False):
+def mark_namespace_for_deletion(user, queues, namespace_gc_queue, force=False):
+  """ Marks a namespace (as referenced by the given user) for deletion. A queue item will be added
+      to delete the namespace's repositories and storage, while the namespace itself will be
+      renamed, disabled, and delinked from other tables.
+  """
+  if not user.enabled:
+    return None
+
   if not force and not user.organization:
     # Ensure that the user is not the sole admin for any organizations. If so, then the user
     # cannot be deleted before those organizations are deleted or reassigned.
@@ -854,10 +861,66 @@ def delete_user(user, queues, force=False):
   for queue in queues:
     queue.delete_namespaced_items(user.username)
 
+  # Delete non-repository related items. This operation is very quick, so we can do so here.
+  _delete_user_linked_data(user)
+
+  with db_transaction():
+    original_username = user.username
+    user = db_for_update(User.select().where(User.id == user.id)).get()
+
+    # Mark the namespace as deleted and ready for GC.
+    try:
+      marker = DeletedNamespace.create(namespace=user,
+                                       original_username=original_username,
+                                       original_email=user.email)
+    except IntegrityError:
+      return
+    
+    # Disable the namespace itself, and replace its various unique fields with UUIDs.
+    user.enabled = False
+    user.username = str(uuid4())
+    user.email = str(uuid4())
+    user.save()
+
+  # Add a queueitem to delete the namespace.
+  marker.queue_id = namespace_gc_queue.put([str(user.id)], json.dumps({
+    'marker_id': marker.id,
+    'original_username': original_username,
+  }))
+  marker.save()
+  return marker.id
+
+
+def delete_namespace_via_marker(marker_id, queues):
+  """ Deletes a namespace referenced by the given DeletedNamespace marker ID. """
+  try:
+    marker = DeletedNamespace.get(id=marker_id)
+  except DeletedNamespace.DoesNotExist:
+    return
+
+  delete_user(marker.namespace, queues)
+
+
+def delete_user(user, queues):
+  """ Deletes a user/organization/robot. Should *not* be called by any user-facing API. Instead,
+      mark_namespace_for_deletion should be used, and the queue should call this method.
+  """
+  # Delete all queue items for the user.
+  for queue in queues:
+    queue.delete_namespaced_items(user.username)
+
   # Delete any repositories under the user's namespace.
   for repo in list(Repository.select().where(Repository.namespace_user == user)):
     repository.purge_repository(user.username, repo.name)
 
+  # Delete non-repository related items.
+  _delete_user_linked_data(user)
+
+  # Delete the user itself.
+  user.delete_instance(recursive=True, delete_nullable=True)
+
+
+def _delete_user_linked_data(user):
   if user.organization:
     # Delete the organization's teams.
     for team in Team.select().where(Team.organization == user):
@@ -879,9 +942,6 @@ def delete_user(user, queues, force=False):
   # falling and only occurs if a superuser is being deleted.
   ServiceKeyApproval.update(approver=None).where(ServiceKeyApproval.approver == user).execute()
 
-  # Delete the user itself.
-  user.delete_instance(recursive=True, delete_nullable=True)
-
 
 def get_pull_credentials(robotname):
   try: