Skip to content

Azure: rework how conflict detection works #147

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
JAVGan wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Azure: rework how conflict detection works #147

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bc4c05e
Azure: rework how conflict detectino works
JAVGan Oct 6, 2025
e09bd83
Azure: wait for active publishing before executing
JAVGan Oct 7, 2025
da1a750
Azure: make wait_for_job_completion retry uniform
JAVGan Oct 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions cloudpub/error.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,5 +21,9 @@ class NotFoundError(ValueError):
"""Represent a missing resource."""


class ConflictError(RuntimeError):
"""Report a submission conflict error."""


class Timeout(Exception):
"""Represent a missing resource."""
25 changes: 11 additions & 14 deletions cloudpub/ms_azure/service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
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
from cloudpub.models.ms_azure import (
RESOURCE_MAPING,
AzureResource,
Expand Down Expand Up @@ -467,31 +467,28 @@ def submit_to_status(
log.debug("Set the status \"%s\" to submission.", status)
return self.configure(resources=cfg_res)

@retry(
wait=wait_fixed(300),
stop=stop_after_delay(max_delay=60 * 60 * 24 * 7), # Give up after retrying for 7 days,
reraise=True,
)
def ensure_can_publish(self, product_id: str) -> None:
"""
Ensure the offer is not already being published.

It will wait for up to 7 days retrying to make sure it's possible to publish before
giving up and raising.
It will raise ConflictError if a publish is already in progress in any submission target.

Args:
product_id (str)
The product ID to check the offer's publishing status
Raises:
RuntimeError: whenever a publishing is already in progress.
ConflictError: whenever a publishing is already in progress for any submission target.
"""
log.info("Ensuring no other publishing jobs are in progress for \"%s\"", product_id)
submission_targets = ["preview", "live"]

for target in submission_targets:
sub = self.get_submission_state(product_id, state=target)
if sub and sub.status and sub.status == "running":
raise RuntimeError(f"The offer {product_id} is already being published to {target}")
for sub in self.get_submissions(product_id):
if sub and sub.status and sub.status != "completed":
msg = (
f"The offer {product_id} is already being published to "
f"{sub.target.targetType}: {sub.status}/{sub.result}"
)
log.error(msg)
raise ConflictError(msg)
Copy link
Collaborator

@lslebodn lslebodn Oct 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we create similar "wait method" as we have for AWS (AWSProductService.wait_active_changesets)?
*

cloudpub/cloudpub/aws/service.py

Lines 283 to 307 in 075853a

def wait_active_changesets(self, entity_id: str) -> None:
"""
Get the first active changeset, if there is one, and wait for it to finish.
Args:
entity_id (str)
The Id of the entity to wait for active changesets
"""
def changeset_not_complete(change_set_list: List[ListChangeSet]) -> bool:
if change_set_list:
self.wait_for_changeset(change_set_list[0].id)
return True
else:
return False
r = Retrying(
stop=stop_after_attempt(self.wait_for_changeset_attempts),
retry=retry_if_result(changeset_not_complete),
)
try:
r(self.get_product_active_changesets, entity_id)
except RetryError:
self._raise_error(Timeout, f"Timed out waiting for {entity_id} to be unlocked")

Copy link
Collaborator Author

@JAVGan JAVGan Oct 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can, yes, but I fear we would end up having some errors like "The submission cannot be pushed to as its not the latest" due to the changes we're attempting to publish not being the latest, or other weird errors like "attempting to remove an image which is already published" due to missing the version which was added in parallel, thus I thought on doing the whole operation again by retrying on pubtools.

Copy link
Collaborator Author

@JAVGan JAVGan Oct 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm thinking here: what I can do, instead, is wait for it before the whole "main" operation starts, like the first thing to do would be wait and then we do everything. Do you think it would be better? I can open a separate PR for that as well 🙂

Copy link
Collaborator Author

@JAVGan JAVGan Oct 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was thinking on opening a different PR but I believe I can reuse this one with a commit on top of it. I had a different idea on how to take advantage of this current implementation in order to implement the "wait" feature.

I'll patch it soon

Copy link
Collaborator Author

@JAVGan JAVGan Oct 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Patched.


def get_plan_tech_config(self, product: Product, plan: PlanSummary) -> VMIPlanTechConfig:
"""
Expand Down
183 changes: 97 additions & 86 deletions tests/ms_azure/test_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,9 @@
from httmock import response
from requests import Response
from requests.exceptions import HTTPError
from tenacity.stop import stop_after_attempt

from cloudpub.common import BaseService
from cloudpub.error import InvalidStateError, NotFoundError
from cloudpub.error import ConflictError, InvalidStateError, NotFoundError
from cloudpub.models.ms_azure import (
ConfigureStatus,
CustomerLeads,
Expand Down Expand Up @@ -750,85 +749,35 @@ def test_submit_to_status_not_found(

mock_configure.assert_not_called()

@pytest.mark.parametrize("target", ["preview", "live"])
@mock.patch("cloudpub.ms_azure.AzureService.get_submission_state")
@mock.patch("cloudpub.ms_azure.AzureService.get_submissions")
def test_ensure_can_publish_success(
self,
mock_getsubst: mock.MagicMock,
target: str,
mock_getsubs: mock.MagicMock,
azure_service: AzureService,
) -> None:
submission = {
"$schema": "https://product-ingestion.azureedge.net/schema/submission/2022-03-01-preview2", # noqa: E501
"id": "submission/ffffffff-ffff-ffff-ffff-ffffffffffff/0",
"product": "product/ffffffff-ffff-ffff-ffff-ffffffffffff",
"target": {"targetType": target},
"lifecycleState": "generallyAvailable",
"status": "completed",
"result": "succeeded",
"created": "2024-07-04T22:06:16.2895521Z",
}
mock_getsubst.return_value = ProductSubmission.from_json(submission)
azure_service.ensure_can_publish.retry.sleep = mock.MagicMock() # type: ignore
azure_service.ensure_can_publish.retry.stop = stop_after_attempt(1) # type: ignore

azure_service.ensure_can_publish("ffffffff-ffff-ffff-ffff-ffffffffffff")

# All targets are called by the method, it should pass all
mock_getsubst.assert_has_calls(
[
mock.call("ffffffff-ffff-ffff-ffff-ffffffffffff", state="preview"),
mock.call("ffffffff-ffff-ffff-ffff-ffffffffffff", state="live"),
]
)

@pytest.mark.parametrize("target", ["preview", "live"])
@mock.patch("cloudpub.ms_azure.AzureService.get_submission_state")
def test_ensure_can_publish_success_after_retry(
self,
mock_getsubst: mock.MagicMock,
target: str,
azure_service: AzureService,
) -> None:
running = {
"$schema": "https://product-ingestion.azureedge.net/schema/submission/2022-03-01-preview2", # noqa: E501
"id": "submission/ffffffff-ffff-ffff-ffff-ffffffffffff/0",
"product": "product/ffffffff-ffff-ffff-ffff-ffffffffffff",
"target": {"targetType": target},
"lifecycleState": "generallyAvailable",
"status": "running",
"result": "pending",
"created": "2024-07-04T22:06:16.2895521Z",
}
complete = {
"$schema": "https://product-ingestion.azureedge.net/schema/submission/2022-03-01-preview2", # noqa: E501
"id": "submission/ffffffff-ffff-ffff-ffff-ffffffffffff/0",
"product": "product/ffffffff-ffff-ffff-ffff-ffffffffffff",
"target": {"targetType": target},
"lifecycleState": "generallyAvailable",
"status": "completed",
"result": "succeeded",
"created": "2024-07-04T22:06:16.2895521Z",
}
mock_getsubst.side_effect = [
ProductSubmission.from_json(running),
ProductSubmission.from_json(running),
ProductSubmission.from_json(complete),
ProductSubmission.from_json(complete),
submissions = [
{
"$schema": "https://product-ingestion.azureedge.net/schema/submission/2022-03-01-preview2", # noqa: E501
"id": "submission/ffffffff-ffff-ffff-ffff-ffffffffffff/0",
"product": "product/ffffffff-ffff-ffff-ffff-ffffffffffff",
"target": {"targetType": tgt},
"lifecycleState": "generallyAvailable",
"status": "completed",
"result": "succeeded",
"created": "2024-07-04T22:06:16.2895521Z",
}
for tgt in ["draft", "preview", "live"]
]
azure_service.ensure_can_publish.retry.sleep = mock.MagicMock() # type: ignore
azure_service.ensure_can_publish.retry.stop = stop_after_attempt(3) # type: ignore
mock_getsubs.return_value = [ProductSubmission.from_json(s) for s in submissions]

azure_service.ensure_can_publish("ffffffff-ffff-ffff-ffff-ffffffffffff")

# Calls for "live" and "preview" for 2 times before success == 4
assert mock_getsubst.call_count == 4
mock_getsubs.assert_called_once()

@pytest.mark.parametrize("target", ["preview", "live"])
@mock.patch("cloudpub.ms_azure.AzureService.get_submission_state")
@mock.patch("cloudpub.ms_azure.AzureService.get_submissions")
def test_ensure_can_publish_raises(
self,
mock_getsubst: mock.MagicMock,
mock_getsubs: mock.MagicMock,
target: str,
azure_service: AzureService,
) -> None:
Expand Down Expand Up @@ -856,17 +805,13 @@ def test_ensure_can_publish_raises(
"result": "pending",
"created": "2024-07-04T22:06:16.2895521Z",
}
if target == "preview":
subs = [ProductSubmission.from_json(sub2), ProductSubmission.from_json(sub1)]
else:
subs = [ProductSubmission.from_json(sub1), ProductSubmission.from_json(sub2)]
mock_getsubst.side_effect = subs
subs = [ProductSubmission.from_json(sub1), ProductSubmission.from_json(sub2)]
mock_getsubs.return_value = subs

err = (
f"The offer ffffffff-ffff-ffff-ffff-ffffffffffff is already being published to {target}"
"The offer ffffffff-ffff-ffff-ffff-ffffffffffff is already being published to "
f"{target}: running/pending"
)
azure_service.ensure_can_publish.retry.sleep = mock.MagicMock() # type: ignore
azure_service.ensure_can_publish.retry.stop = stop_after_attempt(1) # type: ignore

with pytest.raises(RuntimeError, match=err):
azure_service.ensure_can_publish("ffffffff-ffff-ffff-ffff-ffffffffffff")
Expand Down Expand Up @@ -1002,6 +947,74 @@ def test_publish_live_fail_on_retry(
with pytest.raises(RuntimeError, match=expected_err):
azure_service._publish_live(product_obj, "test-product")

@mock.patch("cloudpub.ms_azure.AzureService.compute_targets")
@mock.patch("cloudpub.ms_azure.AzureService.get_productid")
@mock.patch("cloudpub.ms_azure.AzureService.configure")
def test_publish_live_fail_conflict(
self,
mock_configure: mock.MagicMock,
mock_get_productid: mock.MagicMock,
mock_compute_targets: mock.MagicMock,
token: Dict[str, Any],
auth_dict: Dict[str, Any],
configure_success_response: Dict[str, Any],
product: Dict[str, Any],
products_list: Dict[str, Any],
product_summary: Dict[str, Any],
technical_config: Dict[str, Any],
submission: Dict[str, Any],
product_summary_obj: ProductSummary,
plan_summary_obj: PlanSummary,
metadata_azure_obj: mock.MagicMock,
gen2_image: Dict[str, Any],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Ensure operation is aborted when a ConflictError occurs."""
# Prepare testing data
metadata_azure_obj.keepdraft = False
metadata_azure_obj.destination = "example-product/plan-1"
metadata_azure_obj.modular_push = True
mock_get_productid.return_value = "fake-id"
targets = ["preview", "live", "draft"]
mock_compute_targets.return_value = targets

# Set the submission states with conflict on preview
submission_preview = deepcopy(submission)
submission_preview.update(
{"target": {"targetType": "preview"}, "status": "running", "result": "pending"}
)
submission_live = deepcopy(submission)
submission_live.update({"target": {"targetType": "live"}})
mock_configure.return_value = ConfigureStatus.from_json(configure_success_response)

# Expected error
err = (
"The offer ffffffff-ffff-ffff-ffff-ffffffffffff is already being published"
" to preview: running/pending"
)

# Constants
login_url = "https://login.microsoftonline.com/foo/oauth2/token"
base_url = "https://graph.microsoft.com/rp/product-ingestion"
product_id = str(product_summary['id']).split("/")[-1]

# Test
with caplog.at_level(logging.INFO):
with requests_mock.Mocker() as m:
m.post(login_url, json=token)
m.get(f"{base_url}/product", json=products_list)
m.get(f"{base_url}/resource-tree/product/{product_id}", json=product)
m.get(
f"{base_url}/submission/{product_id}",
[
{"json": {"value": [submission, submission_preview, submission_live]}},
],
)
azure_svc = AzureService(auth_dict)

with pytest.raises(ConflictError, match=err):
azure_svc.publish(metadata=metadata_azure_obj)

@mock.patch("cloudpub.ms_azure.AzureService.compute_targets")
@mock.patch("cloudpub.ms_azure.AzureService.get_productid")
@mock.patch("cloudpub.ms_azure.AzureService.configure")
Expand Down Expand Up @@ -1651,12 +1664,14 @@ def test_publish_live_arm64_only(
mock_submit.assert_has_calls(submit_calls)
mock_ensure_publish.assert_called_once_with(product_obj.id)

@mock.patch("cloudpub.ms_azure.AzureService.ensure_can_publish")
@mock.patch("cloudpub.ms_azure.AzureService.compute_targets")
@mock.patch("cloudpub.ms_azure.AzureService.get_productid")
def test_publish_live_when_state_is_preview(
self,
mock_get_productid: mock.MagicMock,
mock_compute_targets: mock.MagicMock,
mock_ensure_publish: mock.MagicMock,
token: Dict[str, Any],
auth_dict: Dict[str, Any],
configure_running_response: Dict[str, Any],
Expand Down Expand Up @@ -1714,8 +1729,6 @@ def test_publish_live_when_state_is_preview(
m.get(
f"{base_url}/submission/{product_id}",
[
{"json": submissions_inprog}, # ensure_can_publish call "preview"
{"json": submissions_inprog}, # ensure_can_publish call "live"
{"json": submissions_inprog}, # _is_submission_in_preview call
{"json": submissions_inprog}, # submit_to_status check prev_state call
{"json": submissions_final}, # submit_to_status validation after configure
Expand Down Expand Up @@ -1749,10 +1762,6 @@ def test_publish_live_when_state_is_preview(
'Requesting the product ID "ffffffff-ffff-ffff-ffff-ffffffffffff" with state "preview".'
in caplog.text
)
assert (
'Ensuring no other publishing jobs are in progress for "ffffffff-ffff-ffff-ffff-ffffffffffff"' # noqa: E501
in caplog.text
)
assert (
'Looking up for submission in state "preview" for "ffffffff-ffff-ffff-ffff-ffffffffffff"' # noqa: E501
in caplog.text
Expand Down Expand Up @@ -1782,7 +1791,9 @@ def test_publish_live_when_state_is_preview(
'Updating the technical configuration for "example-product/plan-1" on "preview".'
not in caplog.text
)
mock_ensure_publish.assert_called_once()

@mock.patch("cloudpub.ms_azure.AzureService.ensure_can_publish")
@mock.patch("cloudpub.ms_azure.AzureService.compute_targets")
@mock.patch("cloudpub.ms_azure.AzureService.get_productid")
@mock.patch("cloudpub.ms_azure.AzureService.configure")
Expand All @@ -1791,6 +1802,7 @@ def test_publish_live_modular_push(
mock_configure: mock.MagicMock,
mock_get_productid: mock.MagicMock,
mock_compute_targets: mock.MagicMock,
mock_ensure_publish: mock.MagicMock,
token: Dict[str, Any],
auth_dict: Dict[str, Any],
configure_success_response: Dict[str, Any],
Expand Down Expand Up @@ -1861,8 +1873,6 @@ def test_publish_live_modular_push(
m.get(
f"{base_url}/submission/{product_id}",
[
{"json": {"value": [submission]}}, # ensure_can_publish call "preview"
{"json": {"value": [submission]}}, # ensure_can_publish call "live"
{"json": {"value": [submission]}}, # push_preview: call submit_status
{"json": {"value": [submission_preview]}}, # push_preview: check result
{"json": {"value": [submission_preview]}}, # push_live: call submit_status
Expand All @@ -1886,6 +1896,7 @@ def test_publish_live_modular_push(
'Performing a modular push to "preview" for "ffffffff-ffff-ffff-ffff-ffffffffffff"'
in caplog.text
)
mock_ensure_publish.assert_called_once()

# Configure request
mock_configure.assert_has_calls(
Expand Down