diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 4af78e2d..518ce61b 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -1,3 +1,28 @@ +4.4.2 - unreleased +------------------ + +## Fixes +* **GCP common-metadata writes now survive concurrent writers.** Labels for + networks, routers, firewalls and key pairs live in the project-wide common + instance metadata, which every write re-uploads under an optimistic + fingerprint. Two things went wrong under parallel use. The retry for a + stale fingerprint only recognised the conflict as an HTTP error, but a + concurrent writer produces it differently: the upload is accepted and the + resulting *operation* completes with ``CONDITION_NOT_MET``, which + ``wait_for_operation`` raised as a plain ``Exception`` the retry ignored. + And - measured in the live suite - GCP accepts several writes against the + same fingerprint while an earlier one is still pending, and a later one can + complete as ``DONE`` with its change absent from the document: 11 of 56 + writes in one run. ``wait_for_operation`` now raises ``GCPOperationError`` + (a ``ProviderInternalException`` carrying the operation's error payload and + ``codes``); a metadata write is only finished once its change reads back, + and is redone on fresh metadata otherwise (``MetadataWriteNotApplied`` once + the retries are exhausted); a write that changes nothing is not sent at + all; ``add_metadata_item`` checks for an existing key itself and raises + ``DuplicateResourceException``, so a retry never appends a second copy; + and ``remove_metadata_item`` returns ``False`` when there was nothing to + remove, as its callers always assumed. + 4.4.1 - August 21, 2026 (sha 093ef669598d9f324be28d400a851396739cf1d8) ---------------------------------------------------------------------- diff --git a/cloudbridge/providers/gcp/helpers.py b/cloudbridge/providers/gcp/helpers.py index 5e98ba4c..d004264b 100644 --- a/cloudbridge/providers/gcp/helpers.py +++ b/cloudbridge/providers/gcp/helpers.py @@ -13,12 +13,45 @@ import tenacity +from cloudbridge.interfaces.exceptions import DuplicateResourceException from cloudbridge.interfaces.exceptions import ProviderInternalException if TYPE_CHECKING: from .provider import GCPCloudProvider +class GCPOperationError(ProviderInternalException): + """A GCP operation completed with an error. + + Raised by ``wait_for_operation`` with the operation's ``error`` payload, + so callers can act on the error codes rather than parse the message. + """ + + def __init__(self, error: Any) -> None: + super().__init__(error) + self.error = error + + @property + def codes(self) -> list[str]: + return [e.get('code', '') for e in self.error.get('errors', [])] + + +class MetadataWriteNotApplied(ProviderInternalException): + """A common-metadata write completed as DONE but is absent when read back. + + Observed under concurrent writers: GCP accepts several writes against + the same fingerprint while an earlier one is still pending, and a later + one can complete without its change surviving. The write is retried on + fresh metadata; this is raised only once the retries are exhausted. + """ + + def __init__(self, keys: list[str]) -> None: + super().__init__( + 'Project metadata write reported done, but these keys do not ' + 'hold the written value: {}'.format(', '.join(keys))) + self.keys = keys + + def gcp_projects(provider: "GCPCloudProvider") -> Any: return provider.gcp_compute.projects() @@ -43,8 +76,18 @@ def get_common_metadata(provider: "GCPCloudProvider") -> Any: return metadata["commonInstanceMetadata"] -def __if_fingerprint_differs(e: BaseException) -> bool: - # return True if the CloudError exception is due to subnet being in use +def __metadata_write_lost_to_a_concurrent_writer(e: BaseException) -> bool: + """Whether ``e`` means a metadata write must be redone on fresh metadata. + + A concurrent writer shows up in three shapes: an HTTP 412 on the request + itself, a submitted operation that completes with ``CONDITION_NOT_MET``, + or - when GCP accepted both writes against the same fingerprint - an + operation that completes as DONE without the change surviving. + """ + if isinstance(e, GCPOperationError): + return 'CONDITION_NOT_MET' in e.codes + if isinstance(e, MetadataWriteNotApplied): + return True if isinstance(e, HttpError): expected_message = 'Supplied fingerprint does not match current ' \ 'metadata fingerprint.' @@ -53,8 +96,14 @@ def __if_fingerprint_differs(e: BaseException) -> bool: return False +def _metadata_values(metadata: Any) -> dict[str, Any]: + return {item['key']: item.get('value') + for item in metadata.get('items', [])} + + @tenacity.retry(stop=tenacity.stop_after_attempt(10), - retry=tenacity.retry_if_exception(__if_fingerprint_differs), + retry=tenacity.retry_if_exception( + __metadata_write_lost_to_a_concurrent_writer), wait=tenacity.wait_exponential(max=10), reraise=True) def gcp_metadata_save_op(provider: "GCPCloudProvider", @@ -66,16 +115,36 @@ def gcp_metadata_save_op(provider: "GCPCloudProvider", retrieves the metadata, invokes the provided callback with that metadata, and saves the metadata using the original fingerprint immediately afterwards, ensuring that update conflicts can be detected. + + The fingerprint alone is not enough: a write that GCP reports as done + can still be missing from the document when another write was accepted + against the same fingerprint. So the write is only finished once the + callback's changes can be read back, and is redone on fresh metadata + otherwise. A callback that changes nothing sends no write at all - every + write re-uploads the whole document and moves the fingerprint, which + would only add contention for other writers. """ def _save_common_metadata(provider: "GCPCloudProvider") -> None: # get the latest metadata (so we get the latest fingerprint) metadata = get_common_metadata(provider) + before = _metadata_values(metadata) # allow callback to do processing on it callback(metadata) + wanted = _metadata_values(metadata) + changes = {key: wanted.get(key) for key in before.keys() | wanted.keys() + if before.get(key) != wanted.get(key)} + if not changes: + return # save the metadata operation = gcp_projects(provider).setCommonInstanceMetadata( project=provider.project_name, body=metadata).execute() provider.wait_for_operation(operation) + # ...and make sure it stuck + current = _metadata_values(get_common_metadata(provider)) + lost = sorted(key for key, value in changes.items() + if current.get(key) != value) + if lost: + raise MetadataWriteNotApplied(lost) # Retry a few times if the fingerprints conflict _save_common_metadata(provider) @@ -98,15 +167,24 @@ def _update_metadata_key(metadata: Any) -> None: gcp_metadata_save_op(provider, _update_metadata_key) -# This function will raise an HttpError with message containing -# "Metadata has duplicate key" if it's not unique, unlike the previous -# method which either adds or updates the value corresponding to that key def add_metadata_item(provider: "GCPCloudProvider", key: str, value: str) -> None: + """Add ``key``, raising ``DuplicateResourceException`` if it is taken. + + Unlike ``modify_or_add_metadata_item`` this never overwrites. The + check is done on the freshly fetched metadata so that a retry after a + lost write does not append a second copy of a key that did land: a key + already holding this very value is this write, and counts as done. + """ def _add_metadata_key(metadata: Any) -> None: - entry = {'key': key, 'value': value} entries = metadata.get('items', []) - entries.append(entry) + existing = [item for item in entries if item['key'] == key] + if existing: + if existing[-1].get('value') == value: + return + raise DuplicateResourceException( + 'Metadata key {0} already exists'.format(key)) + entries.append({'key': key, 'value': value}) # Reassign explicitly in case the original get returned [] although # if not it will be already updated metadata['items'] = entries @@ -136,30 +214,25 @@ def get_metadata_item_value(provider: "GCPCloudProvider", key: str) -> Any: def remove_metadata_item(provider: "GCPCloudProvider", key: str) -> bool: - def _remove_metadata_by_key(metadata: Any) -> bool | None: - items = metadata.get('items', []) - # No metadata to delete - if not items: - return False - else: - entries = [item for item in metadata.get('items', []) - if item['key'] != key] - - # Make sure only one entry is deleted - if len(entries) < len(items) - 1: - raise ProviderInternalException("Multiple metadata entries " - "found for the same key {}" - .format(key)) - # If none is deleted indicate so by returning False - elif len(entries) == len(items): - return False + """Remove ``key``; ``False`` if there was nothing to remove.""" + removed = False - else: - metadata['items'] = entries - return None + def _remove_metadata_by_key(metadata: Any) -> None: + nonlocal removed + items = metadata.get('items', []) + entries = [item for item in items if item['key'] != key] + # Make sure only one entry is deleted + if len(entries) < len(items) - 1: + raise ProviderInternalException("Multiple metadata entries " + "found for the same key {}" + .format(key)) + # A retry after a lost write finds the key still present and + # removes it again; the first attempt decides whether it was there. + removed = removed or len(entries) < len(items) + metadata['items'] = entries gcp_metadata_save_op(provider, _remove_metadata_by_key) - return True + return removed def __if_label_fingerprint_differs(e: BaseException) -> bool: diff --git a/cloudbridge/providers/gcp/provider.py b/cloudbridge/providers/gcp/provider.py index 33d97eab..5976050f 100644 --- a/cloudbridge/providers/gcp/provider.py +++ b/cloudbridge/providers/gcp/provider.py @@ -30,6 +30,7 @@ from cloudbridge.interfaces.services import SecurityService from cloudbridge.interfaces.services import StorageService +from .helpers import GCPOperationError from .services import GCPComputeService from .services import GCPDnsService from .services import GCPNetworkingService @@ -401,7 +402,7 @@ def wait_for_operation(self, operation: Any, region: str | None = None, result = operations.get(**args).execute() if result['status'] == 'DONE': if 'error' in result: - raise Exception(result['error']) + raise GCPOperationError(result['error']) return result time.sleep(0.5) diff --git a/cloudbridge/providers/gcp/services.py b/cloudbridge/providers/gcp/services.py index 1bb4fc19..f0a4a357 100644 --- a/cloudbridge/providers/gcp/services.py +++ b/cloudbridge/providers/gcp/services.py @@ -188,7 +188,12 @@ def create(self, name: str, GCPKeyPair.KP_TAG_PREFIX + name, metadata_value) return GCPKeyPair(provider, kp_info, private_key) + except DuplicateResourceException: + raise DuplicateResourceException( + 'A KeyPair with name {0} already exists'.format(name)) except googleapiclient.errors.HttpError as err: + # GCP's own duplicate-key rejection, should a concurrent create + # slip past the check on the fetched metadata. if err.resp.get('content-type', '').startswith('application/json'): message = (json.loads(err.content).get('error', {}) .get('errors', [{}])[0].get('message')) diff --git a/tests/test_gcp_metadata_save.py b/tests/test_gcp_metadata_save.py new file mode 100644 index 00000000..32b2f4cb --- /dev/null +++ b/tests/test_gcp_metadata_save.py @@ -0,0 +1,208 @@ +"""Making GCP common-metadata writes stick. + +GCP keeps labels and key pairs in the project-wide common instance metadata, +which every write re-uploads under an optimistic fingerprint. Under +concurrent writers two things happen that a write has to survive: the +operation fails with ``CONDITION_NOT_MET`` (a different path from the +HTTP-level 412), and - observed in the live suite - the operation reports +``DONE`` yet the change is absent from the document afterwards. Both must +lead to the write being re-applied on freshly fetched metadata. No SDK is +involved: the compute client is a fake with a server-side document, and the +provider's real ``wait_for_operation`` polls it. +""" + +import unittest +from unittest import mock + +import tenacity + +from cloudbridge.interfaces.exceptions import DuplicateResourceException +from cloudbridge.providers.gcp.helpers import GCPOperationError +from cloudbridge.providers.gcp.helpers import MetadataWriteNotApplied +from cloudbridge.providers.gcp.helpers import add_metadata_item +from cloudbridge.providers.gcp.helpers import gcp_metadata_save_op +from cloudbridge.providers.gcp.helpers import modify_or_add_metadata_item +from cloudbridge.providers.gcp.helpers import remove_metadata_item +from cloudbridge.providers.gcp.provider import GCPCloudProvider + +FINGERPRINT_CONFLICT = { + 'errors': [{'code': 'CONDITION_NOT_MET', + 'message': 'Supplied fingerprint does not match current ' + 'metadata fingerprint.'}]} +OTHER_FAILURE = { + 'errors': [{'code': 'RESOURCE_NOT_FOUND', + 'message': "The resource 'projects/p' was not found"}]} + +# Scripted outcomes for successive set operations. +APPLY = 'apply' # the write lands and the fingerprint moves on +LOST = 'lost' # reported DONE, but nothing changed +CONFLICT = 'conflict' # CONDITION_NOT_MET, nothing changed + + +class _Call: + def __init__(self, result): + self._result = result + + def execute(self): + return self._result + + +class _FakeCompute: + """A compute client holding one project metadata document. + + ``projects().get()`` returns a copy of the document; each + ``setCommonInstanceMetadata()`` yields an operation whose fate is the + next entry in ``outcomes``, resolved when ``wait_for_operation`` polls + ``globalOperations().get()``. + """ + + def __init__(self, outcomes, items=None): + self.outcomes = list(outcomes) + self.document = {'fingerprint': 'fp-0', + 'items': list(items or [])} + self.version = 0 + self.fetches = 0 + self.saved_bodies = [] + self._pending = {} + + def projects(self): + return self + + def globalOperations(self): + return self + + def get(self, project, operation=None): + if operation is not None: + return self._resolve(operation) + self.fetches += 1 + return _Call({'commonInstanceMetadata': { + 'fingerprint': self.document['fingerprint'], + 'items': [dict(i) for i in self.document['items']]}}) + + def setCommonInstanceMetadata(self, project, body): + self.saved_bodies.append(body) + name = 'op-%d' % len(self.saved_bodies) + self._pending[name] = body + return _Call({'name': name}) + + def _resolve(self, name): + body = self._pending.pop(name) + outcome = self.outcomes.pop(0) + result = {'status': 'DONE'} + if outcome == APPLY: + self.version += 1 + self.document = {'fingerprint': 'fp-%d' % self.version, + 'items': [dict(i) for i in body['items']]} + elif outcome == CONFLICT: + result['error'] = FINGERPRINT_CONFLICT + elif outcome == LOST: + pass + else: + result['error'] = outcome + return _Call(result) + + def keys(self): + return {i['key']: i['value'] for i in self.document['items']} + + +class _FakeProvider: + project_name = 'p' + wait_for_operation = GCPCloudProvider.wait_for_operation + + def __init__(self, outcomes, items=None): + self.gcp_compute = _FakeCompute(outcomes, items) + + +class GCPMetadataSaveTestCase(unittest.TestCase): + + def setUp(self): + # The production wait between attempts is exponential backoff; these + # tests are about whether a retry happens, not how long it waits. + patcher = mock.patch.object(gcp_metadata_save_op.retry, 'wait', + tenacity.wait_none()) + patcher.start() + self.addCleanup(patcher.stop) + + def test_fingerprint_conflict_is_retried_with_fresh_metadata(self): + provider = _FakeProvider([CONFLICT, APPLY]) + callback = mock.Mock(side_effect=lambda md: md['items'].append( + {'key': 'k', 'value': 'v'})) + + gcp_metadata_save_op(provider, callback) + + # Two attempts, each on metadata fetched anew so the retry carries + # the fingerprint the conflict invalidated. + self.assertEqual(callback.call_count, 2) + self.assertEqual( + [b['fingerprint'] for b in provider.gcp_compute.saved_bodies], + ['fp-0', 'fp-0']) + self.assertEqual(provider.gcp_compute.keys(), {'k': 'v'}) + + def test_write_reported_done_but_absent_is_reapplied(self): + # The live suite showed operations completing as DONE with the + # change missing from the document. A write is only finished once + # it can be read back. + provider = _FakeProvider([LOST, APPLY]) + + modify_or_add_metadata_item(provider, 'k', 'v') + + self.assertEqual(len(provider.gcp_compute.saved_bodies), 2) + self.assertEqual(provider.gcp_compute.keys(), {'k': 'v'}) + + def test_write_that_never_lands_is_reported_after_retries(self): + provider = _FakeProvider([LOST] * 10) + + with self.assertRaises(MetadataWriteNotApplied) as raised: + modify_or_add_metadata_item(provider, 'k', 'v') + + self.assertIn('k', str(raised.exception)) + self.assertEqual(len(provider.gcp_compute.saved_bodies), 10) + + def test_other_operation_failures_are_raised_as_typed_errors(self): + provider = _FakeProvider([OTHER_FAILURE]) + callback = mock.Mock(side_effect=lambda md: md['items'].append( + {'key': 'k', 'value': 'v'})) + + with self.assertRaises(GCPOperationError) as raised: + gcp_metadata_save_op(provider, callback) + + self.assertEqual(callback.call_count, 1) + self.assertEqual(raised.exception.codes, ['RESOURCE_NOT_FOUND']) + self.assertIn("was not found", str(raised.exception)) + + def test_a_write_that_changes_nothing_is_not_sent(self): + # Every set re-uploads the whole document and moves the fingerprint, + # so a no-op (removing an absent key, re-setting the current value) + # would only add contention for other writers. + provider = _FakeProvider([], items=[{'key': 'k', 'value': 'v'}]) + + self.assertFalse(remove_metadata_item(provider, 'absent')) + modify_or_add_metadata_item(provider, 'k', 'v') + + self.assertEqual(provider.gcp_compute.saved_bodies, []) + + def test_remove_is_reapplied_until_the_key_is_gone(self): + provider = _FakeProvider([LOST, APPLY], + items=[{'key': 'k', 'value': 'v'}]) + + self.assertTrue(remove_metadata_item(provider, 'k')) + + self.assertEqual(len(provider.gcp_compute.saved_bodies), 2) + self.assertEqual(provider.gcp_compute.keys(), {}) + + def test_add_is_satisfied_by_its_own_value_already_present(self): + # A retry must not append a second copy if the earlier attempt did + # land after all. + provider = _FakeProvider([], items=[{'key': 'k', 'value': 'v'}]) + + add_metadata_item(provider, 'k', 'v') + + self.assertEqual(provider.gcp_compute.saved_bodies, []) + + def test_add_rejects_a_key_someone_else_holds(self): + provider = _FakeProvider([], items=[{'key': 'k', 'value': 'theirs'}]) + + with self.assertRaises(DuplicateResourceException): + add_metadata_item(provider, 'k', 'mine') + + self.assertEqual(provider.gcp_compute.saved_bodies, []) diff --git a/tox.ini b/tox.ini index db826cb3..7bfdcbc0 100644 --- a/tox.ini +++ b/tox.ini @@ -16,14 +16,16 @@ commands = # see pyproject.toml for coverage options; setup.cfg for flake8 # inferring it from gaps between result lines in the CI log, which # cannot separate a slow test from a worker waiting for work. coverage run --source=cloudbridge -m pytest -v --durations=25 {posargs:-n 5 tests/} - # Emit any CB_TEST_TRACE output (see tests/conftest.py). A no-op - # when tracing is off, which is the default. - python -c "import glob,sys;[sys.stdout.write(open(f).read()) for f in sorted(glob.glob('cb-trace-*.log'))]" # Combine parallel-mode data files and emit Cobertura XML for upload # by coverallsapp/github-action in CI. Locally this produces # coverage.xml in the project root, which IDEs can also consume. coverage combine coverage xml +# Emit any CB_TEST_TRACE output (see tests/conftest.py). A no-op when tracing +# is off, which is the default. This lives in commands_post rather than after +# pytest in commands because tox stops at the first failing command, and a +# failing run is the one whose trace is worth reading. +commands_post = python -c "import glob,sys;[sys.stdout.write(open(f).read()) for f in sorted(glob.glob('cb-trace-*.log'))]" setenv = # Fix for moto import issue: https://github.com/travis-ci/travis-ci/issues/7940 BOTO_CONFIG=/dev/null