migrate to server side-apply

Author: nammn

docker/mongodb-kubernetes-tests/kubeobject/customobject.py

@@ -1,11 +1,11 @@
 from __future__ import annotations
 
-import copy
 from datetime import datetime, timedelta
 from typing import Dict, Optional
 
 import yaml
 from kubernetes import client
+from kubernetes.dynamic import DynamicClient
 
 
 class CustomObject:
@@ -73,8 +73,14 @@ def __init__(
         # Last time this object was updated
         self.last_update: datetime = None
 
+        # Store the api_client for later use
+        self._api_client = api_client if api_client else client.ApiClient()
+
         # Sets the API used for this particular type of object
-        self.api = client.CustomObjectsApi(api_client=api_client)
+        self.api = client.CustomObjectsApi(api_client=self._api_client)
+
+        # Dynamic client for server-side apply
+        self._dynamic_client = DynamicClient(self._api_client)
 
         if not hasattr(self, "backing_obj"):
             self.backing_obj = {
@@ -112,21 +118,30 @@ def update(self) -> CustomObject:
         """Updates the object in Kubernetes. Deleting keys is done by setting them to None"""
         return create_or_update(self)
 
-    def patch(self) -> CustomObject:
-        """Patch the object in Kubernetes. Deleting keys is done by setting them to None"""
-        obj = self.api.patch_namespaced_custom_object(
-            self.group,
-            self.version,
-            self.namespace,
-            self.plural,
-            self.name,
-            self.backing_obj,
+    def patch(self, field_manager: str = "e2e-tests") -> CustomObject:
+        """Apply the object in Kubernetes using server-side apply.
+
+        Server-side apply (SSA) provides proper field ownership tracking and
+        conflict resolution, avoiding race conditions when both the operator
+        and tests modify the same CR.
+
+        Args:
+            field_manager: Identifier for the field manager (owner of the fields being applied)
+        """
+        api_version = self.backing_obj.get("apiVersion", f"{self.group}/{self.version}")
+        kind = self.backing_obj.get("kind", self.kind)
+
+        resource_api = self._dynamic_client.resources.get(api_version=api_version, kind=kind)
+        result = resource_api.server_side_apply(
+            namespace=self.namespace,
+            body=self.backing_obj,
+            field_manager=field_manager,
         )
-        self.backing_obj = obj
+        self.backing_obj = result.to_dict()
 
         self._register_updated()
 
-        return obj
+        return self
 
     def _register_updated(self):
         """Register the last time the object was updated from Kubernetes."""
@@ -316,49 +331,21 @@ def get_crd_names(
 def create_or_update(resource: CustomObject) -> CustomObject:
     """
     Tries to create the resource. If resource already exists (resulting in 409 Conflict),
-    then it updates it instead. If the resource has been modified externally (operator)
-    we try to do a client-side merge/override
+    then it updates it instead using server-side apply.
+
+    Server-side apply (SSA) with force=True handles conflicts automatically by taking
+    ownership of conflicting fields, eliminating the need for client-side merge retries.
     """
-    tries = 0
     if not resource.bound:
         try:
             resource.create()
         except client.ApiException as e:
             if e.status != 409:
                 raise e
+            # Resource already exists, use SSA to update it
             resource.patch()
     else:
-        while tries < 10:
-            if tries > 0:  # The first try we don't need to do client-side merge apply
-                # do a client-side-apply
-                new_back_obj_to_apply = copy.deepcopy(resource.backing_obj)  # resource and changes we want to apply
-
-                resource.load()  # resource from the server overwrites resource.backing_obj
-
-                # Merge annotations, and labels.
-                # Client resource takes precedence
-                # Spec from the given resource is taken,
-                # since the operator is not supposed to do changes to the spec.
-                # There can be cases where the obj from the server does not contain annotations/labels, but the object
-                # we want to apply has them. But that is highly unlikely, and we can add that code in case that happens.
-                resource["spec"] = new_back_obj_to_apply["spec"]
-                if "metadata" in resource and "annotations" in resource["metadata"]:
-                    resource["metadata"]["annotations"].update(new_back_obj_to_apply["metadata"]["annotations"])
-                if "metadata" in resource and "labels" in resource["metadata"]:
-                    resource["metadata"]["labels"].update(new_back_obj_to_apply["metadata"]["labels"])
-            try:
-                resource.patch()
-                break
-            except client.ApiException as e:
-                if e.status != 409:
-                    raise e
-                print(
-                    "detected a resource conflict. That means the operator applied a change "
-                    "to the same resource we are trying to change"
-                    "Applying a client-side merge!"
-                )
-                tries += 1
-                if tries == 10:
-                    raise Exception("Tried client side merge 10 times and did not succeed")
+        # SSA with force=True handles conflicts automatically
+        resource.patch()
 
     return resource

docker/mongodb-kubernetes-tests/kubeobject/kubeobject.py

@@ -11,6 +11,7 @@
 from kubeobject.exceptions import ObjectNotBoundException
 from kubernetes import client
 from kubernetes.client.api import ApiextensionsV1Api, CustomObjectsApi
+from kubernetes.dynamic import DynamicClient
 
 
 class KubeObject(object):
@@ -42,11 +43,14 @@ def init_attributes(self):
         # This is the object that will contain the definition of the Custom Object when bound
         self.__dict__[KubeObject.BACKING_OBJ] = Box(default_box=True)
 
+        # Store the API client for reuse
+        self.__dict__["_api_client"] = client.ApiClient()
+
         # Set an API to work with.
-        # TODO: Allow for a better experience; api could be defined from env variables,
-        # in_cluster or whatever. See if this is needed. Can we run our samples with
-        # in_cluster, or based on different clusters pointed at by env variables?
-        self.__dict__["api"] = CustomObjectsApi()
+        self.__dict__["api"] = CustomObjectsApi(api_client=self.__dict__["_api_client"])
+
+        # Dynamic client for server-side apply
+        self.__dict__["_dynamic_client"] = DynamicClient(self.__dict__["_api_client"])
 
         # Set `auto_reload` to `True` if it needs to be reloaded before every
         # read of an attribute. This considers the `auto_reload_period`
@@ -86,19 +90,32 @@ def read(self, name: str, namespace: str):
         self._register_update()
         return self
 
-    def update(self):
+    def update(self, field_manager: str = "e2e-tests"):
+        """Updates the object in Kubernetes using server-side apply.
+
+        Server-side apply (SSA) provides proper field ownership tracking and
+        conflict resolution, avoiding race conditions when both the operator
+        and tests modify the same CR.
+
+        Args:
+            field_manager: Identifier for the field manager (owner of the fields being applied)
+        """
         if not self.bound:
             # there's no corresponding object in the Kubernetes cluster
             raise ObjectNotBoundException
 
-        obj = self.api.patch_namespaced_custom_object(
-            name=self.name,
+        body = self.__dict__[KubeObject.BACKING_OBJ].to_dict()
+        api_version = body.get("apiVersion", f"{self.crd['group']}/{self.crd['version']}")
+        kind = body.get("kind")
+
+        resource_api = self.__dict__["_dynamic_client"].resources.get(api_version=api_version, kind=kind)
+        result = resource_api.server_side_apply(
             namespace=self.namespace,
-            **self.crd,
-            body=self.__dict__[KubeObject.BACKING_OBJ].to_dict(),
+            body=body,
+            field_manager=field_manager,
         )
 
-        self.__dict__[KubeObject.BACKING_OBJ] = Box(obj, default_box=True)
+        self.__dict__[KubeObject.BACKING_OBJ] = Box(result.to_dict(), default_box=True)
         self._register_update()
 
         return self

docker/mongodb-kubernetes-tests/kubeobject/test_custom_object.py

@@ -32,27 +32,55 @@
     anotherAttrStr: value1
 """
 
+# Shared storage for mock data
+_stored_body = []
 
-def mocked_custom_api():
-    stored_body = []
 
+class MockResourceResult:
+    """Mock for DynamicClient resource result."""
+
+    def __init__(self, data):
+        self._data = data
+
+    def to_dict(self):
+        return self._data
+
+
+class MockResource:
+    """Mock for DynamicClient resource API."""
+
+    def server_side_apply(self, namespace, body, field_manager):
+        _stored_body.append(body)
+        return MockResourceResult(body)
+
+
+class MockDynamicClient:
+    """Mock for DynamicClient."""
+
+    class Resources:
+        def get(self, api_version, kind):
+            return MockResource()
+
+    resources = Resources()
+
+
+def mocked_dynamic_client():
+    return MockDynamicClient()
+
+
+def mocked_custom_api():
     def get_namespaced_custom_object(group, version, namespace, plural, name):
-        if len(stored_body) > 0:
-            return stored_body[-1]
+        if len(_stored_body) > 0:
+            return _stored_body[-1]
         return {"name": name}
 
     def create_namespaced_custom_object(group, version, namespace, plural, body: dict):
         body.update({"name": body["metadata"]["name"]})
-        stored_body.append(body)
-        return body
-
-    def patch_namespaced_custom_object(group, version, namespace, plural, name, body: dict):
-        stored_body.append(body)
+        _stored_body.append(body)
         return body
 
     base = MagicMock()
     base.get_namespaced_custom_object = MagicMock(side_effect=get_namespaced_custom_object)
-    base.patch_namespaced_custom_object = MagicMock(side_effect=patch_namespaced_custom_object)
     base.create_namespaced_custom_object = MagicMock(side_effect=create_namespaced_custom_object)
 
     return base
@@ -68,10 +96,13 @@ def mocked_crd_return_value():
     )
 
 
+@mock.patch("kubeobject.customobject.DynamicClient")
 @mock.patch("kubeobject.customobject.client.CustomObjectsApi")
 @mock.patch("kubeobject.customobject.get_crd_names", return_value=mocked_crd_return_value())
-def test_custom_object_creation(mocked_get_crd_names, mocked_client):
+def test_custom_object_creation(mocked_get_crd_names, mocked_client, mocked_dynamic):
+    _stored_body.clear()
     mocked_client.return_value = mocked_custom_api()
+    mocked_dynamic.return_value = mocked_dynamic_client()
     custom = CustomObject(
         "my-dummy-object",
         "my-dummy-namespace",
@@ -84,10 +115,13 @@ def test_custom_object_creation(mocked_get_crd_names, mocked_client):
     assert custom["name"] == "my-dummy-object"
 
 
+@mock.patch("kubeobject.customobject.DynamicClient")
 @mock.patch("kubeobject.customobject.client.CustomObjectsApi")
 @mock.patch("kubeobject.customobject.get_crd_names", return_value=mocked_crd_return_value())
-def test_custom_object_read_from_disk(mocked_get_crd_names, mocked_client):
+def test_custom_object_read_from_disk(mocked_get_crd_names, mocked_client, mocked_dynamic):
+    _stored_body.clear()
     mocked_client.return_value = mocked_custom_api()
+    mocked_dynamic.return_value = mocked_dynamic_client()
     with mock.patch(
         "kubeobject.customobject.open",
         mock.mock_open(read_data=yaml_data0),
@@ -106,10 +140,13 @@ def test_custom_object_read_from_disk(mocked_get_crd_names, mocked_client):
         assert custom["spec"]["subDoc"]["anotherAttrStr"] == "value1"
 
 
+@mock.patch("kubeobject.customobject.DynamicClient")
 @mock.patch("kubeobject.customobject.client.CustomObjectsApi")
 @mock.patch("kubeobject.customobject.get_crd_names", return_value=mocked_crd_return_value())
-def test_custom_object_read_from_disk_with_dat_from_yaml(mocked_get_crd_names, mocked_client):
+def test_custom_object_read_from_disk_with_dat_from_yaml(mocked_get_crd_names, mocked_client, mocked_dynamic):
+    _stored_body.clear()
     mocked_client.return_value = mocked_custom_api()
+    mocked_dynamic.return_value = mocked_dynamic_client()
     with mock.patch(
         "kubeobject.customobject.open",
         mock.mock_open(read_data=yaml_data0),
@@ -137,10 +174,13 @@ def test_custom_object_read_from_disk_with_dat_from_yaml(mocked_get_crd_names, m
     assert custom.plural == "dummies"
 
 
+@mock.patch("kubeobject.customobject.DynamicClient")
 @mock.patch("kubeobject.customobject.client.CustomObjectsApi")
 @mock.patch("kubeobject.customobject.get_crd_names", return_value=mocked_crd_return_value())
-def test_custom_object_can_be_subclassed_create(mocked_crd_return_value, mocked_client):
+def test_custom_object_can_be_subclassed_create(mocked_crd_return_value, mocked_client, mocked_dynamic):
+    _stored_body.clear()
     mocked_client.return_value = mocked_custom_api()
+    mocked_dynamic.return_value = mocked_dynamic_client()
 
     class Subklass(CustomObject):
         pass
@@ -156,10 +196,13 @@ class Subklass(CustomObject):
     assert a.__class__.__name__ == "Subklass"
 
 
+@mock.patch("kubeobject.customobject.DynamicClient")
 @mock.patch("kubeobject.customobject.client.CustomObjectsApi")
 @mock.patch("kubeobject.customobject.get_crd_names", return_value=mocked_crd_return_value())
-def test_custom_object_can_be_subclassed_from_yaml(mocked_crd_return_value, mocked_client):
+def test_custom_object_can_be_subclassed_from_yaml(mocked_crd_return_value, mocked_client, mocked_dynamic):
+    _stored_body.clear()
     mocked_client.return_value = mocked_custom_api()
+    mocked_dynamic.return_value = mocked_dynamic_client()
 
     class Subklass(CustomObject):
         pass
@@ -175,10 +218,13 @@ class Subklass(CustomObject):
         assert a.__class__.__name__ == "Subklass"
 
 
+@mock.patch("kubeobject.customobject.DynamicClient")
 @mock.patch("kubeobject.customobject.client.CustomObjectsApi")
 @mock.patch("kubeobject.customobject.get_crd_names", return_value=mocked_crd_return_value())
-def test_custom_object_defined(mocked_crd_return_value, mocked_client):
+def test_custom_object_defined(mocked_crd_return_value, mocked_client, mocked_dynamic):
+    _stored_body.clear()
     mocked_client.return_value = mocked_custom_api()
+    mocked_dynamic.return_value = mocked_dynamic_client()
     klass = CustomObject.define("Dummy", plural="dummies", group="dummy.com", version="v1")
 
     k = klass("my-dummy", "default").create()
@@ -189,9 +235,12 @@ def test_custom_object_defined(mocked_crd_return_value, mocked_client):
     assert repr(k) == "Dummy('my-dummy', 'default')"
 
 
+@mock.patch("kubeobject.customobject.DynamicClient")
 @mock.patch("kubeobject.customobject.client.CustomObjectsApi")
-def test_defined_wont_require_api_if_all_parameteres_are_provided(mocked_client):
+def test_defined_wont_require_api_if_all_parameteres_are_provided(mocked_client, mocked_dynamic):
+    _stored_body.clear()
     mocked_client.return_value = mocked_custom_api()
+    mocked_dynamic.return_value = mocked_dynamic_client()
     BaseKlass = CustomObject.define("Dummy", kind="Dummy", plural="dummies", group="dummy.com", version="v1")
 
     class SubKlass(BaseKlass):
@@ -207,11 +256,14 @@ def get_spec(self):
     assert k.get_spec() == {"testAttr": "value"}
 
 
+@mock.patch("kubeobject.customobject.DynamicClient")
 @mock.patch("kubeobject.customobject.client.CustomObjectsApi")
 @mock.patch("kubeobject.customobject.get_crd_names", return_value=mocked_crd_return_value())
-def test_custom_object_auto_reload(mocked_get_crd_names, mocked_client):
+def test_custom_object_auto_reload(mocked_get_crd_names, mocked_client, mocked_dynamic):
+    _stored_body.clear()
     instance = mocked_custom_api()
     mocked_client.return_value = instance
+    mocked_dynamic.return_value = mocked_dynamic_client()
 
     klass = CustomObject.define("Dummy", plural="dummies", group="dummy.com", version="v1")
     k = klass("my-dummy", "default")
@@ -265,9 +317,13 @@ def test_raises_if_no_namespace():
             CustomObject.from_yaml("some-other-file.yaml", name="some-name")
 
 
+@mock.patch("kubeobject.customobject.DynamicClient")
+@mock.patch("kubeobject.customobject.client.CustomObjectsApi")
 @mock.patch("kubeobject.customobject.get_crd_names", return_value=mocked_crd_return_value())
-def test_name_is_set_as_argument(_):
+def test_name_is_set_as_argument(_, mocked_client, mocked_dynamic):
     # TODO: what is this test supposed to do?
+    mocked_client.return_value = mocked_custom_api()
+    mocked_dynamic.return_value = mocked_dynamic_client()
     with mock.patch(
         "kubeobject.customobject.open",
         mock.mock_open(read_data=yaml_data1),