Hotfix/SS-1807 patch inconsistent deployment (#478)

Author: hamzaimran08

apps/helpers.py

@@ -312,8 +312,15 @@ def create_instance_from_form(form, project, app_slug, app_id=None, force_redepl
     assert project is not None, "This function requires a project object"
 
     new_app = app_id is None
+    requested_app_slug = app_slug
 
-    logger.debug(f"Creating or updating a user app via UI form for app_id={app_id}, new_app={new_app}")
+    logger.info(
+        "create_instance_from_form.start app_id=%s new_app=%s app_slug=%s project_id=%s",
+        app_id,
+        new_app,
+        app_slug,
+        project.pk,
+    )
 
     if new_app:
         do_deploy = True
@@ -348,6 +355,12 @@ def create_instance_from_form(form, project, app_slug, app_id=None, force_redepl
                     break
 
     subdomain_name, is_created_by_user = get_subdomain_name(form)
+    logger.info(
+        "create_instance_from_form.subdomain_selected app_id=%s subdomain=%s is_created_by_user=%s",
+        app_id,
+        subdomain_name,
+        is_created_by_user,
+    )
 
     instance = form.save(commit=False)
 
@@ -361,11 +374,23 @@ def create_instance_from_form(form, project, app_slug, app_id=None, force_redepl
     subdomain = Subdomain.objects.get(subdomain=subdomain_name, project=project, is_created_by_user=is_created_by_user)
     assert subdomain is not None
     assert subdomain.subdomain == subdomain_name
+    logger.info(
+        "create_instance_from_form.subdomain_ready app_id=%s subdomain=%s created=%s",
+        app_id,
+        subdomain_name,
+        created,
+    )
 
     if not new_app:
         handle_subdomain_change(instance, subdomain, subdomain_name)
 
     app_slug = handle_shiny_proxy_case(instance, app_slug, app_id)
+    logger.info(
+        "create_instance_from_form.app_slug_resolved app_id=%s requested_slug=%s resolved_slug=%s",
+        app_id,
+        requested_app_slug,
+        app_slug,
+    )
 
     app = get_app(app_slug)
 
@@ -378,12 +403,37 @@ def create_instance_from_form(form, project, app_slug, app_id=None, force_redepl
 
     setup_instance(instance, subdomain, app, project, user_action)
     instance_id = save_instance_and_related_data(instance, form)
+    logger.info(
+        "create_instance_from_form.instance_saved app_id=%s instance_id=%s user_action=%s do_deploy=%s",
+        app_id,
+        instance_id,
+        user_action,
+        do_deploy,
+    )
 
     if do_deploy:
-        logger.debug(f"Now deploying resource app with app_id = {app_id}")
-        deploy_resource.delay(instance.serialize())
+        serialized_instance = instance.serialize()
+        logger.info(
+            "create_instance_from_form.enqueue_on_commit app_id=%s instance_id=%s model=%s pk=%s",
+            app_id,
+            instance_id,
+            serialized_instance.get("model"),
+            serialized_instance.get("pk"),
+        )
+
+        def enqueue_deploy_task():
+            logger.info(
+                "create_instance_from_form.enqueue_dispatch app_id=%s instance_id=%s model=%s pk=%s",
+                app_id,
+                instance_id,
+                serialized_instance.get("model"),
+                serialized_instance.get("pk"),
+            )
+            deploy_resource.delay(serialized_instance)
+
+        transaction.on_commit(enqueue_deploy_task)
     else:
-        logger.debug(f"Not re-deploying this app with app_id = {app_id}")
+        logger.info("create_instance_from_form.deploy_skipped app_id=%s instance_id=%s", app_id, instance_id)
 
     if waffle.switch_is_active("doi_minting_using_invenio"):
         image_value_changed = False
@@ -485,10 +535,16 @@ def handle_subdomain_change(instance: Any, subdomain: str, subdomain_name: str)
     from .tasks import delete_resource
 
     assert instance is not None, "instance is required"
+    logger.info(
+        "handle_subdomain_change.start instance_id=%s current_subdomain=%s requested_subdomain=%s",
+        instance.pk,
+        instance.subdomain.subdomain if instance.subdomain else None,
+        subdomain_name,
+    )
 
     if instance.subdomain is None:
         # The subdomain is not yet created, nothing to do
-        logger.debug("The subdomain is not yet created, nothing to do")
+        logger.info("handle_subdomain_change.skip_no_existing_subdomain instance_id=%s", instance.pk)
         return
 
     if instance.subdomain.subdomain != subdomain_name:
@@ -498,8 +554,21 @@ def handle_subdomain_change(instance: Any, subdomain: str, subdomain_name: str)
         old_subdomain = instance.subdomain
         instance.subdomain = subdomain
         instance.save(update_fields=["subdomain"])
+        logger.info(
+            "handle_subdomain_change.updated instance_id=%s old_subdomain=%s new_subdomain=%s",
+            instance.pk,
+            old_subdomain.subdomain if old_subdomain else None,
+            subdomain_name,
+        )
         if old_subdomain and not old_subdomain.is_created_by_user:
             old_subdomain.delete()
+            logger.info(
+                "handle_subdomain_change.deleted_old_subdomain instance_id=%s old_subdomain=%s",
+                instance.pk,
+                old_subdomain.subdomain,
+            )
+    else:
+        logger.info("handle_subdomain_change.no_change instance_id=%s subdomain=%s", instance.pk, subdomain_name)
 
 
 def handle_shiny_proxy_case(instance, app_slug, app_id):
@@ -526,6 +595,14 @@ def setup_instance(instance, subdomain, app, project, user_action=None, is_creat
     instance.project = project
     instance.owner = project.owner
     instance.latest_user_action = user_action
+    logger.info(
+        "setup_instance.assigned instance_id=%s subdomain=%s app_slug=%s project_id=%s user_action=%s",
+        instance.pk,
+        subdomain.subdomain if subdomain else None,
+        app.slug if app else None,
+        project.pk if project else None,
+        user_action,
+    )
 
 
 def save_instance_and_related_data(instance: Any, form: Any) -> int:
@@ -535,13 +612,15 @@ def save_instance_and_related_data(instance: Any, form: Any) -> int:
     Returns:
     - int: The Id of the new or updated app instance.
     """
+    logger.info("save_instance_and_related_data.start instance_id=%s", instance.pk)
     instance.save()
     form.save_m2m()
     instance.set_k8s_values()
     instance.url = get_URI(instance)
     # For MLFLOW, we need to set the k8s_values again to update the URL
     instance.set_k8s_values()
     instance.save(update_fields=["k8s_values", "url"])
+    logger.info("save_instance_and_related_data.finish instance_id=%s url=%s", instance.id, instance.url)
     return instance.id
 
 

apps/tasks.py

@@ -24,6 +24,25 @@
 logger = get_logger(__name__)
 
 CHART_REGEX = re.compile(r"^(?P<chart>.+):(?P<version>.+)$")
+DEPLOY_RESOURCE_MAX_RETRIES = 3
+DEPLOY_RESOURCE_RETRY_BASE_SECONDS = 10
+DEPLOY_RESOURCE_RETRY_MAX_SECONDS = 30
+
+
+class MissingSerializedInstanceError(ValueError):
+    """Raised when a serialized model/pk cannot be resolved from the database."""
+
+    def __init__(self, model: str, pk: int, base_instance_exists: bool):
+        self.model = model
+        self.pk = pk
+        self.base_instance_exists = base_instance_exists
+        super().__init__(
+            f"No instance found for model {model} with pk {pk} (base_instance_exists={base_instance_exists})"
+        )
+
+
+def _retry_countdown(current_retries: int) -> int:
+    return min(DEPLOY_RESOURCE_RETRY_BASE_SECONDS * (2**current_retries), DEPLOY_RESOURCE_RETRY_MAX_SECONDS)
 
 
 @app.task
@@ -198,11 +217,71 @@ def get_manifest_yaml(release_name: str, namespace: str = "default") -> tuple[st
         return e.stdout, e.stderr
 
 
-@shared_task
+@shared_task(bind=True, max_retries=DEPLOY_RESOURCE_MAX_RETRIES)
 @transaction.atomic
-def deploy_resource(serialized_instance):
-    instance: BaseAppInstance = deserialize(serialized_instance)
-    logger.info("Deploying resource for instance %s", instance)
+def deploy_resource(self, serialized_instance):
+    model = serialized_instance.get("model") if isinstance(serialized_instance, dict) else None
+    pk = serialized_instance.get("pk") if isinstance(serialized_instance, dict) else None
+    task_id = getattr(self.request, "id", None)
+
+    logger.info(
+        "deploy_resource.start task_id=%s model=%s pk=%s retry=%s",
+        task_id,
+        model,
+        pk,
+        self.request.retries,
+    )
+
+    try:
+        instance: BaseAppInstance = deserialize(serialized_instance)
+    except MissingSerializedInstanceError as exc:
+        retries = self.request.retries
+        if retries < self.max_retries:
+            countdown = _retry_countdown(retries)
+            logger.warning(
+                "deploy_resource.missing_instance_retry task_id=%s model=%s pk=%s retry=%s/%s "
+                "countdown=%ss base_instance_exists=%s",
+                task_id,
+                exc.model,
+                exc.pk,
+                retries + 1,
+                self.max_retries,
+                countdown,
+                exc.base_instance_exists,
+            )
+            raise self.retry(exc=exc, countdown=countdown)
+
+        logger.error(
+            "deploy_resource.missing_instance_exhausted task_id=%s model=%s pk=%s retries=%s "
+            "base_instance_exists=%s",
+            task_id,
+            exc.model,
+            exc.pk,
+            retries,
+            exc.base_instance_exists,
+        )
+        raise
+
+    logger.info(
+        "deploy_resource.instance_resolved task_id=%s model=%s pk=%s instance_id=%s app_slug=%s",
+        task_id,
+        model,
+        pk,
+        instance.pk,
+        instance.app.slug,
+    )
+
+    deleted_on = getattr(instance, "deleted_on", None)
+    if instance.latest_user_action in {"Deleting", "SystemDeleting"} or deleted_on is not None:
+        logger.info(
+            "deploy_resource.skip_deleting task_id=%s instance_id=%s latest_user_action=%s deleted_on=%s",
+            task_id,
+            instance.pk,
+            instance.latest_user_action,
+            deleted_on,
+        )
+        return
+
     values = get_merged_k8s_values(instance, ensure_up_to_date=True)
     release = values["subdomain"]
     chart: str = instance.chart
@@ -263,17 +342,59 @@ def deploy_resource(serialized_instance):
         if not valid_deployment:
             logger.warning(f"The deployment manifest file is INVALID for release {release}. {validation_output}")
 
+    logger.info(
+        "deploy_resource.helm_install_start task_id=%s instance_id=%s release=%s namespace=%s chart=%s version=%s",
+        task_id,
+        instance.pk,
+        release,
+        values["namespace"],
+        chart,
+        version,
+    )
+
     # Install the app using Helm install
     output, error = helm_install(release, chart, values["namespace"], values_file, version)
     success = not error
 
+    if not success:
+        retries = self.request.retries
+        logger.warning(
+            "deploy_resource.helm_install_failed task_id=%s instance_id=%s retry=%s/%s release=%s stderr=%s",
+            task_id,
+            instance.pk,
+            retries,
+            self.max_retries,
+            release,
+            error,
+        )
+        if retries < self.max_retries:
+            countdown = _retry_countdown(retries)
+            logger.info(
+                "deploy_resource.helm_install_retry task_id=%s instance_id=%s retry=%s/%s countdown=%ss",
+                task_id,
+                instance.pk,
+                retries + 1,
+                self.max_retries,
+                countdown,
+            )
+            raise self.retry(exc=RuntimeError(error or "Helm install failed"), countdown=countdown)
+
+    logger.info(
+        "deploy_resource.helm_install_done task_id=%s instance_id=%s success=%s release=%s",
+        task_id,
+        instance.pk,
+        success,
+        release,
+    )
+
     helm_info = {"success": success, "info": {"stdout": output, "stderr": error}}
 
     instance.info = dict(helm=helm_info)
     # instance.app_status.status = "Created" if success else "Failed"
 
     # Only update the info field to avoid overriding other modified fields elsewhere
     instance.save(update_fields=["info"])
+    logger.info("deploy_resource.info_saved task_id=%s instance_id=%s success=%s", task_id, instance.pk, success)
 
     # In development, also generate and validate the k8s deployment manifest
     if settings.DEBUG:
@@ -287,6 +408,14 @@ def deploy_resource(serialized_instance):
         if deployment_file:
             subprocess.run(["rm", "-f", deployment_file])
 
+    logger.info(
+        "deploy_resource.finish task_id=%s instance_id=%s success=%s valid_deployment=%s",
+        task_id,
+        instance.pk,
+        success,
+        valid_deployment,
+    )
+
 
 @shared_task
 @transaction.atomic
@@ -303,7 +432,13 @@ def delete_resource(serialized_instance, initiated_by_str: str):
     - serialized_instance: A serialized version of the app to be deleted.
     - initiated_by_str: A string of enum AppActionOrigin indicating the source of the deletion (user|system).
     """
-    logger.debug(f"Type of serialized_instance is {type(serialized_instance)}")
+    logger.info(
+        "delete_resource.start model=%s pk=%s initiated_by=%s payload_type=%s",
+        serialized_instance.get("model") if isinstance(serialized_instance, dict) else None,
+        serialized_instance.get("pk") if isinstance(serialized_instance, dict) else None,
+        initiated_by_str,
+        type(serialized_instance),
+    )
 
     initiated_by = AppActionOrigin(initiated_by_str)
     assert initiated_by == AppActionOrigin.USER or initiated_by == AppActionOrigin.SYSTEM
@@ -362,12 +497,14 @@ def deserialize(serialized_instance):
 
         model_class = apps.get_model(app_label, model_name)
         instance = model_class.objects.get(pk=pk)
+        logger.info("deserialize.resolved model=%s pk=%s concrete_model=%s", model, pk, model_class.__name__)
 
         return instance
     except (KeyError, ValueError) as e:
         raise ValueError(f"Invalid serialized data format: {e}")
     except ObjectDoesNotExist:
-        raise ValueError(f"No instance found for model {model} with pk {pk}")
+        base_instance_exists = BaseAppInstance.objects.filter(pk=pk).exists()
+        raise MissingSerializedInstanceError(model=model, pk=pk, base_instance_exists=base_instance_exists)
 
 
 @app.task