Merge pull request #652 from nolar/deployments-and-replicasets

Evade collision of annotations in Deployments & ReplicaSets

kopf/storage/conventions.py

@@ -33,12 +33,48 @@
 import base64
 import hashlib
 import warnings
-from typing import Any, Collection, Iterable, Set
+from typing import Any, Collection, Iterable, Optional, Set
 
 from kopf.structs import bodies, patches
 
 
-class StorageKeyFormingConvention:
+class CollisionEvadingConvention:
+    """
+    A helper mixin to evade collisions in annotations propagated down by K8s.
+
+    For some resources, such as ReplicaSets owned by Deployments,
+    the annotations are implicitly propagated by Kubernetes
+    from the owning resources down to the owned resources.
+
+    As a result, if both resources are served by Kopf-based operators with
+    the same or default identity, the owner's annotations overwrite those
+    of the resource, which causes all kinds of chaos when e.g. the diff-base
+    mismatches the resource's schema or the handlers' progress is miscalculated.
+
+    To evade this, Kopf adds special marks to all annotations of all resources
+    known to be overwritten by Kubernetes -- in order to preserve the state
+    regardless of whether the parent's annotations are already propagated:
+    this can happen much later when the owning resource is started to be served
+    hours, days, months after the owned resource has stored its state.
+
+    The only known case at the moment is caused by this behaviour in Kubernetes:
+
+    * https://github.com/kubernetes/kubernetes/blob/v1.20.2/pkg/controller/deployment/util/deployment_util.go#L230-L234
+    * https://github.com/kubernetes/kubernetes/blob/v1.20.2/pkg/controller/deployment/util/deployment_util.go#L310-L341
+
+    We assume this does not happen to other resources unless proven otherwise.
+    """
+
+    def mark_key(self, key: str, *, body: bodies.Body) -> str:
+        owners = body.meta.get('ownerReferences', [])
+        kind = body.get('kind')
+        if kind == 'ReplicaSet' and any(owner['kind'] == 'Deployment' for owner in owners):
+            return f"{key}-ofDRS"  # no need to generalise for a single known case
+        else:
+            return key
+
+
+class StorageKeyFormingConvention(CollisionEvadingConvention):
     """
     A helper mixin to manage annotations/labels naming as per K8s restrictions.
 
@@ -107,7 +143,8 @@ def __init__(
         if len(self.prefix or '') > 253 - 63 - 1:
             warnings.warn("The annotations prefix is too long. It can cause errors when PATCHing.")
 
-    def make_keys(self, key: str) -> Iterable[str]:
+    def make_keys(self, key: str, *, body: Optional[bodies.Body] = None) -> Iterable[str]:
+        key = key if body is None else self.mark_key(key, body=body)
         v2_keys = [self.make_v2_key(key)]
         v1_keys = [self.make_v1_key(key)] if self.v1 else []
         return v2_keys + list(set(v1_keys) - set(v2_keys))

kopf/storage/diffbase.py

@@ -128,7 +128,7 @@ def build(
     ) -> bodies.BodyEssence:
         essence = super().build(body=body, extra_fields=extra_fields)
         annotations = essence.get('metadata', {}).get('annotations', {})
-        for full_key in self.make_keys(self.key):
+        for full_key in self.make_keys(self.key, body=body):
             if full_key in annotations:
                 del annotations[full_key]
         return essence
@@ -138,7 +138,7 @@ def fetch(
             *,
             body: bodies.Body,
     ) -> Optional[bodies.BodyEssence]:
-        for full_key in self.make_keys(self.key):
+        for full_key in self.make_keys(self.key, body=body):
             encoded = body.metadata.annotations.get(full_key, None)
             decoded = json.loads(encoded) if encoded is not None else None
             if decoded is not None:
@@ -154,7 +154,7 @@ def store(
     ) -> None:
         encoded: str = json.dumps(essence, separators=(',', ':'))  # NB: no spaces
         encoded += '\n'  # for better kubectl presentation without wrapping (same as kubectl's one)
-        for full_key in self.make_keys(self.key):
+        for full_key in self.make_keys(self.key, body=body):
             patch.metadata.annotations[full_key] = encoded
         self._store_marker(prefix=self.prefix, patch=patch, body=body)
 

kopf/storage/progress.py

@@ -180,7 +180,7 @@ def fetch(
             key: handlers.HandlerId,
             body: bodies.Body,
     ) -> Optional[ProgressRecord]:
-        for full_key in self.make_keys(key):
+        for full_key in self.make_keys(key, body=body):
             key_field = ['metadata', 'annotations', full_key]
             encoded = dicts.resolve(body, key_field, None)
             decoded = json.loads(encoded) if encoded is not None else None
@@ -198,7 +198,7 @@ def store(
     ) -> None:
         decoded = {key: val for key, val in record.items() if self.verbose or val is not None}
         encoded = json.dumps(decoded, separators=(',', ':'))  # NB: no spaces
-        for full_key in self.make_keys(key):
+        for full_key in self.make_keys(key, body=body):
             key_field = ['metadata', 'annotations', full_key]
             dicts.ensure(patch, key_field, encoded)
         self._store_marker(prefix=self.prefix, patch=patch, body=body)
@@ -211,7 +211,7 @@ def purge(
             patch: patches.Patch,
     ) -> None:
         absent = object()
-        for full_key in self.make_keys(key):
+        for full_key in self.make_keys(key, body=body):
             key_field = ['metadata', 'annotations', full_key]
             body_value = dicts.resolve(body, key_field, absent)
             patch_value = dicts.resolve(patch, key_field, absent)
@@ -227,7 +227,7 @@ def touch(
             patch: patches.Patch,
             value: Optional[str],
     ) -> None:
-        for full_key in self.make_keys(self.touch_key):
+        for full_key in self.make_keys(self.touch_key, body=body):
             key_field = ['metadata', 'annotations', full_key]
             body_value = dicts.resolve(body, key_field, None)
             if body_value != value:  # also covers absent-vs-None cases.

tests/persistence/test_annotations_hashing.py

@@ -89,6 +89,19 @@ def test_keys_deduplication(cls):
     assert v2_key in keys
 
 
+@pytest.mark.parametrize('kind, owners, expected', [
+    pytest.param('ReplicaSet', [{'kind': 'Deployment'}], 'kopf.dev/xyz-ofDRS', id='DRS'),
+    pytest.param('ReplicaSet', [{'kind': 'OtherOwner'}], 'kopf.dev/xyz', id='not-deployment'),
+    pytest.param('OtherKind', [{'kind': 'Deployment'}], 'kopf.dev/xyz', id='not-replicaset'),
+])
+@pytest.mark.parametrize('cls', STORAGE_KEY_FORMING_CLASSES)
+def test_keys_of_replicaset_owned_by_deployment(cls, kind, owners, expected):
+    storage = cls(v1=True, prefix='kopf.dev')
+    body = Body({'kind': kind, 'metadata': {'ownerReferences': owners}})
+    keys = storage.make_keys('xyz', body=body)
+    assert set(keys) == {expected}
+
+
 @pytest.mark.parametrize('prefix, provided_key, expected_key', COMMON_KEYS + V1_KEYS)
 @pytest.mark.parametrize('cls', STORAGE_KEY_FORMING_CLASSES)
 def test_key_hashing_v1(cls, prefix, provided_key, expected_key):