Merge pull request #349 from CloudVE/gcp-metadata-conflict-retry

Retry GCP metadata writes on an operation-level fingerprint conflict

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)
 ----------------------------------------------------------------------
 

cloudbridge/providers/gcp/helpers.py

@@ -13,12 +13,45 @@ from googleapiclient.errors import HttpError
 
 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 modify_or_add_metadata_item(provider: "GCPCloudProvider", key: str,
     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:

cloudbridge/providers/gcp/provider.py

@@ -30,6 +30,7 @@ from cloudbridge.interfaces.services import NetworkingService
 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 @@ class GCPCloudProvider(BaseCloudProvider):
             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)

cloudbridge/providers/gcp/services.py

@@ -188,7 +188,12 @@ class GCPKeyPairService(BaseKeyPairService):
                                       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'))

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, [])

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