Azure: Retry on publishing when a conflict occurs

This commit changes the Azure module to retry publishing the VM image
whenever a submission in progress/conflict happens.

It will first attempt to change the target to `preview` or `live` for 3
times and then, if the exception comes as `ConflictError` or
`RunningSubmissionError` it will restart the publishing task.

Assisted-By: Cursor
Signed-off-by: Jonathan Gangi <[email protected]>

Author: JAVGan

cloudpub/ms_azure/service.py

@@ -6,13 +6,13 @@
 
 from deepdiff import DeepDiff
 from requests import HTTPError
-from tenacity import retry
-from tenacity.retry import retry_if_result
+from tenacity import retry, wait_exponential
+from tenacity.retry import retry_if_exception_type, retry_if_result
 from tenacity.stop import stop_after_attempt, stop_after_delay
 from tenacity.wait import wait_chain, wait_fixed
 
 from cloudpub.common import BaseService
-from cloudpub.error import InvalidStateError, NotFoundError
+from cloudpub.error import ConflictError, InvalidStateError, NotFoundError, RunningSubmissionError
 from cloudpub.models.ms_azure import (
     RESOURCE_MAPING,
     AzureResource,
@@ -38,6 +38,8 @@
 from cloudpub.ms_azure.session import PartnerPortalSession
 from cloudpub.ms_azure.utils import (
     AzurePublishingMetadata,
+    check_for_conflict,
+    check_for_running_submission,
     create_disk_version_from_scratch,
     is_azure_job_not_complete,
     is_sas_present,
@@ -175,6 +177,10 @@ def _wait_for_job_completion(self, job_id: str) -> ConfigureStatus:
         job_details = self._query_job_details(job_id=job_id)
         if job_details.job_result == "failed":
             error_message = f"Job {job_id} failed: \n{job_details.errors}"
+            if check_for_conflict(job_details):
+                self._raise_error(ConflictError, error_message)
+            elif check_for_running_submission(job_details):
+                self._raise_error(RunningSubmissionError, error_message)
             self._raise_error(InvalidStateError, error_message)
         elif job_details.job_result == "succeeded":
             log.debug(f"Job {job_id} succeeded")
@@ -548,7 +554,7 @@ def _publish_preview(self, product: Product, product_name: str) -> None:
             if res.job_result != 'succeeded' or not self.get_submission_state(
                 product.id, state="preview"
             ):
-                errors = "\n".join(res.errors)
+                errors = "\n".join(["%s: %s" % (error.code, error.message) for error in res.errors])
                 failure_msg = (
                     f"Failed to submit the product {product.id} to preview. "
                     f"Status: {res.job_result} Errors: {errors}"
@@ -576,13 +582,19 @@ def _publish_live(self, product: Product, product_name: str) -> None:
         res = self.submit_to_status(product_id=product.id, status='live')
 
         if res.job_result != 'succeeded' or not self.get_submission_state(product.id, state="live"):
-            errors = "\n".join(res.errors)
+            errors = "\n".join(["%s: %s" % (error.code, error.message) for error in res.errors])
             failure_msg = (
                 f"Failed to submit the product {product.id} to live. "
                 f"Status: {res.job_result} Errors: {errors}"
             )
             raise RuntimeError(failure_msg)
 
+    @retry(
+        retry=retry_if_exception_type([ConflictError, RunningSubmissionError]),
+        wait=wait_exponential(multiplier=1, min=60, max=60 * 60 * 24 * 7),
+        stop=stop_after_attempt(3),
+        reraise=True,
+    )
     def publish(self, metadata: AzurePublishingMetadata) -> None:
         """
         Associate a VM image with a given product listing (destination) and publish it if required.

cloudpub/ms_azure/utils.py

@@ -1,5 +1,6 @@
 # SPDX-License-Identifier: GPL-3.0-or-later
 import logging
+import re
 from operator import attrgetter
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -592,3 +593,22 @@ def logdiff(diff: DeepDiff) -> None:
     """Log the offer diff if it exists."""
     if diff:
         log.warning(f"Found the following offer diff before publishing:\n{diff.pretty()}")
+
+
+def check_for_conflict(job_details: ConfigureStatus) -> bool:
+    """Check if the job details contain a conflict error."""
+    err_lookup = r"The submission cannot be pushed to \w+ as its not the latest .*"
+    for error in job_details.errors:
+        if error.code == "conflict" and re.match(err_lookup, error.message):
+            return True
+    return False
+
+
+def check_for_running_submission(job_details: ConfigureStatus) -> bool:
+    """Check if the job details contain a running submission error."""
+    err_lookup = r"An In Progress submission [0-9]+ already exists."
+    for error in job_details.errors:
+        err_msg = error.message
+        if error.code == "internalServerError" and re.match(err_lookup, err_msg):
+            return True
+    return False