Merge ~pappacena/turnip:repo-async-creation-status into turnip:master

[Thiago F. Pappacena]

Reporting status on repositories that are currently being created

Unmerged commits

5f18244... by Thiago F. Pappacena on 2021-03-31

Reporting status on repositories that are currently being created

diff --git a/config.yaml b/config.yaml
index 0273840..e216a3a 100644
--- a/config.yaml
+++ b/config.yaml
@@ -21,6 +21,7 @@ openid_provider_root: https://testopenid.test/
 site_name: git.launchpad.test
 main_site_root: https://launchpad.test/
 celery_broker: pyamqp://guest@localhost//
+celery_backend: file:///var/tmp/git.launchpad.test
 statsd_host: 127.0.0.1
 statsd_port: 8125
 statsd_prefix: turnip
diff --git a/turnip/api/store.py b/turnip/api/store.py
index f2719a7..ca3b194 100644
--- a/turnip/api/store.py
+++ b/turnip/api/store.py
@@ -3,6 +3,7 @@
 
 import base64
 import itertools
+import json
 import logging
 import os
 import re
@@ -295,7 +296,8 @@ class AlreadyExistsError(GitError):
 
 
 def init_repo(repo_path, clone_from=None, clone_refs=False,
-              alternate_repo_paths=None, is_bare=True, log=None):
+              alternate_repo_paths=None, is_bare=True, log=None,
+              metadata=None):
     """Initialise a new git repository or clone from existing."""
     if os.path.exists(repo_path):
         raise AlreadyExistsError(repo_path)
@@ -305,7 +307,7 @@ def init_repo(repo_path, clone_from=None, clone_refs=False,
     init_repository(repo_path, is_bare)
 
     log.info("Running set_repository_creating(%s, True)" % repo_path)
-    set_repository_creating(repo_path, True)
+    set_repository_creating(repo_path, True, metadata)
 
     if clone_from:
         # The clone_from's objects and refs are in fact cloned into a
@@ -354,31 +356,41 @@ def init_repo(repo_path, clone_from=None, clone_refs=False,
     ensure_config(repo_path)  # set repository configuration defaults
 
     log.info("Running set_repository_creating(%s, False)" % repo_path)
+    # XXX: pappacena: Remove this. It's here just to pretend a fork takes
+    # time while testing it locally.
+    import time; time.sleep(30)
     set_repository_creating(repo_path, False)
 
 
[e-mail]
-def init_and_confirm_repo(untranslated_path, repo_path, clone_from=None,
[e-mail](bind=True)
+def init_and_confirm_repo(self, untranslated_path, repo_path, clone_from=None,
                           clone_refs=False, alternate_repo_paths=None,
                           is_bare=True):
-    logger = tasks_logger
-    xmlrpc_endpoint = config.get("virtinfo_endpoint")
-    xmlrpc_timeout = float(config.get("virtinfo_timeout"))
-    xmlrpc_auth_params = {"user": "+launchpad-services"}
-    xmlrpc_proxy = TimeoutServerProxy(
-        xmlrpc_endpoint, timeout=xmlrpc_timeout, allow_none=True)
+    self.update_state(state='RUNNING')
+    try:
+        logger = tasks_logger
+        xmlrpc_endpoint = config.get("virtinfo_endpoint")
+        xmlrpc_timeout = float(config.get("virtinfo_timeout"))
+        xmlrpc_auth_params = {"user": "+launchpad-services"}
+        xmlrpc_proxy = TimeoutServerProxy(
+            xmlrpc_endpoint, timeout=xmlrpc_timeout, allow_none=True)
+    except:
+        self.update_state(state="FAILED")
+        raise
     try:
         logger.info(
             "Initializing and confirming repository creation: "
             "%s; %s; %s; %s; %s", repo_path, clone_from, clone_refs,
             alternate_repo_paths, is_bare)
         init_repo(
-            repo_path, clone_from, clone_refs, alternate_repo_paths, is_bare)
+            repo_path, clone_from, clone_refs, alternate_repo_paths, is_bare,
+            metadata={"task_id": self.request.id})
         logger.debug(
             "Confirming repository creation: %s; %s",
             untranslated_path, xmlrpc_auth_params)
         xmlrpc_proxy.confirmRepoCreation(untranslated_path, xmlrpc_auth_params)
     except Exception as e:
+        self.update_state(state='FAILED')
         logger.error("Error creating repository at %s: %s", repo_path, e)
         try:
             delete_repo(repo_path)
@@ -388,6 +400,8 @@ def init_and_confirm_repo(untranslated_path, repo_path, clone_from=None,
             "Aborting repository creation: %s; %s",
             untranslated_path, xmlrpc_auth_params)
         xmlrpc_proxy.abortRepoCreation(untranslated_path, xmlrpc_auth_params)
+    finally:
+        self.update_state(state='SUCCESS')
 
 
 @contextmanager
@@ -435,17 +449,26 @@ def get_default_branch(repo_path):
     return repo.references['HEAD'].target
 
 
-def set_repository_creating(repo_path, is_creating):
+def set_repository_creating(repo_path, is_creating, metadata=None):
+    """Set the repository state as "creating" or "not creating".
+
+    If is_creating is True, metadata will be saved in creation status file.
+    """
     if not os.path.exists(repo_path):
         raise ValueError("Repository %s does not exist." % repo_path)
     file_path = os.path.join(repo_path, REPOSITORY_CREATING_FILE_NAME)
     if is_creating:
-        open(file_path, 'a').close()
+        with open(file_path, 'w+') as fd:
+            fd.write(json.dumps(metadata or {}))
     else:
         try:
             os.unlink(file_path)
         except OSError:
             pass
+        # Remove the AsyncResult creation object from Celery backend.
+        result = get_creation_async_result(repo_path)
+        if result:
+            result.forget()
 
 
 def is_repository_available(repo_path):
@@ -458,6 +481,37 @@ def is_repository_available(repo_path):
     return not os.path.exists(status_file_path)
 
 
+def get_creation_async_result(repo_path):
+    status_file_path = os.path.join(repo_path, REPOSITORY_CREATING_FILE_NAME)
+    if not os.path.exists(status_file_path):
+        return None
+    with open(status_file_path) as fd:
+        metadata = fd.read()
+        if not metadata:
+            return None
+        task_id = json.loads(metadata).get("task_id")
+        return app.AsyncResult(task_id)
+
+
+def get_repository_creation_status(repo_path):
+    """
+    For repositories that are currently beign created though
+    `init_and_confirm_repo`, this method returns the current status of the
+    creation.
+
+    :return: None if the repository is not being created; Otherwise, one of
+    the following values:
+        RUNNING - The creation is still in progress.
+        FAILED - The creation attempt failed.
+        SUCCESS - The creating finished successfully
+    """
+    if is_repository_available(repo_path):
+        return None
+
+    result = get_creation_async_result(repo_path)
+    return result.status if result else None
+
+
 def set_default_branch(repo_path, target):
     repo = Repository(repo_path)
     repo.set_head(target)
diff --git a/turnip/api/views.py b/turnip/api/views.py
index cda3a00..242599f 100644
--- a/turnip/api/views.py
+++ b/turnip/api/views.py
@@ -96,7 +96,8 @@ class RepoAPI(BaseAPI):
             raise exc.HTTPNotFound()
         return {
             'default_branch': store.get_default_branch(repo_path),
-            'is_available': store.is_repository_available(repo_path)
+            'is_available': store.is_repository_available(repo_path),
+            'creation_status': store.get_repository_creation_status(repo_path)
             }
 
     def _patch_default_branch(self, repo_path, value):
diff --git a/turnip/tasks.py b/turnip/tasks.py
index 1f4131a..33612cc 100644
--- a/turnip/tasks.py
+++ b/turnip/tasks.py
@@ -14,7 +14,8 @@ from celery.utils.log import get_task_logger
 from turnip.config import config
 
 
-app = Celery('tasks', broker=config.get('celery_broker'))
+app = Celery('tasks', broker=config.get('celery_broker'),
+             backend=config.get('celery_backend'))
 app.conf.update(imports=('turnip.api.store', ))
 
 logger = get_task_logger(__name__)