mgr/rook: fix mds/fs create/apply/update

Signed-off-by: Sage Weil <sage@redhat.com>

diff --git a/src/pybind/mgr/rook/module.py b/src/pybind/mgr/rook/module.py
index ecd27701cc20ecf1213d0c41ac12c0b7c2e6d5c1..0408fee7061d2ce7416d9a4037b41da5a2c140a6 100644 (file)
--- a/src/pybind/mgr/rook/module.py
+++ b/src/pybind/mgr/rook/module.py
@@ -4,7 +4,7 @@ import functools
 import os
 
 from ceph.deployment import inventory
-from ceph.deployment.service_spec import ServiceSpec, NFSServiceSpec, RGWSpec
+from ceph.deployment.service_spec import ServiceSpec, NFSServiceSpec, RGWSpec, PlacementSpec
 
 try:
     from typing import List, Dict, Optional, Callable, Any
@@ -264,9 +264,9 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
         spec = {}
         spec['mon'] = orchestrator.ServiceDescription(
             service_name='mon',
-            spec=orchestrator.ServiceSpec(
+            spec=ServiceSpec(
                 'mon',
-                placement=orchestrator.PlacementSpec(
+                placement=PlacementSpec(
                     count=cl['spec'].get('mon', {}).get('count', 1),
                 ),
             ),
@@ -276,9 +276,9 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
         )
         spec['mgr'] = orchestrator.ServiceDescription(
             service_name='mgr',
-            spec=orchestrator.ServiceSpec(
+            spec=ServiceSpec(
                 'mgr',
-                placement=orchestrator.PlacementSpec.from_string('count:1'),
+                placement=PlacementSpec.from_string('count:1'),
             ),
             size=1,
             container_image_name=image_name,
@@ -287,9 +287,9 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
         if not cl['spec'].get('crashCollector', {}).get('disable', False):
             spec['crash'] = orchestrator.ServiceDescription(
                 service_name='crash',
-                spec=orchestrator.ServiceSpec(
+                spec=ServiceSpec(
                     'crash',
-                    placement=orchestrator.PlacementSpec.from_string('all:true'),
+                    placement=PlacementSpec.from_string('all:true'),
                 ),
                 size=num_nodes,
                 container_image_name=image_name,
@@ -311,9 +311,9 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
                 total_mds = active * 2
             spec[svc] = orchestrator.ServiceDescription(
                 service_name=svc,
-                spec=orchestrator.ServiceSpec(
+                spec=ServiceSpec(
                     svc,
-                    placement=orchestrator.PlacementSpec(count=active),
+                    placement=PlacementSpec(count=active),
                 ),
                 size=total_mds,
                 container_image_name=image_name,
@@ -377,11 +377,6 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
             mgr=self
         )
 
-    def add_mds(self, spec):
-        # type: (ServiceSpec) -> RookCompletion
-        return self._service_add_decorate('MDS', spec,
-                                       self.rook_cluster.add_filesystem)
-
     def add_rgw(self, spec):
         # type: (RGWSpec) -> RookCompletion
         return self._service_add_decorate('RGW', spec,
@@ -427,12 +422,8 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
 
     def apply_mds(self, spec):
         # type: (ServiceSpec) -> RookCompletion
-        num = spec.placement.count
-        return write_completion(
-            lambda: self.rook_cluster.update_mds_count(spec.service_id, num),
-            "Updating MDS server count in {0} to {1}".format(spec.service_id, num),
-            mgr=self
-        )
+        return self._service_add_decorate('MDS', spec,
+                                          self.rook_cluster.apply_filesystem)
 
     def apply_nfs(self, spec):
         # type: (NFSServiceSpec) -> RookCompletion

diff --git a/src/pybind/mgr/rook/rook_cluster.py b/src/pybind/mgr/rook/rook_cluster.py
index 57aec4e7828ff9cdc211c9de04d9bdd749b8dd71..e704d78fc022e014d891b4b020c1a8bf5e31388b 100644 (file)
--- a/src/pybind/mgr/rook/rook_cluster.py
+++ b/src/pybind/mgr/rook/rook_cluster.py
@@ -363,28 +363,34 @@ class RookCluster(object):
             else:
                 raise
 
-    def add_filesystem(self, spec):
+    def apply_filesystem(self, spec):
         # type: (ServiceSpec) -> None
         # TODO use spec.placement
         # TODO warn if spec.extended has entries we don't kow how
         #      to action.
+        def _update_fs(current, new):
+            # type: (cfs.CephFilesystem, cfs.CephFilesystem) -> cfs.CephFilesystem
+            new.spec.metadataServer.activeCount = spec.placement.count or 1
+            return new
 
-        rook_fs = cfs.CephFilesystem(
-            apiVersion=self.rook_env.api_name,
-            metadata=dict(
-                name=spec.service_id,
-                namespace=self.rook_env.namespace,
-            ),
-            spec=cfs.Spec(
-                metadataServer=cfs.MetadataServer(
-                    activeCount=spec.placement.count,
-                    activeStandby=True
+        def _create_fs():
+            # type: () -> cfs.CephFilesystem
+            return cfs.CephFilesystem(
+                apiVersion=self.rook_env.api_name,
+                metadata=dict(
+                    name=spec.service_id,
+                    namespace=self.rook_env.namespace,
+                ),
+                spec=cfs.Spec(
+                    metadataServer=cfs.MetadataServer(
+                        activeCount=spec.placement.count or 1,
+                        activeStandby=True
+                    )
                 )
             )
-        )
-
-        with self.ignore_409("CephFilesystem '{0}' already exists".format(spec.service_id)):
-            self.rook_api_post("cephfilesystems/", body=rook_fs.to_json())
+        return self._create_or_patch(
+            cfs.CephFilesystem, 'cephfilesystems', spec.service_id,
+            _update_fs, _create_fs)
 
     def add_nfsgw(self, spec):
         # TODO use spec.placement
@@ -477,13 +483,6 @@ class RookCluster(object):
             return new
         return self._patch(ccl.CephCluster, 'cephclusters', self.rook_env.cluster_name, _update_mon_count)
 
-    def update_mds_count(self, svc_id, newcount):
-        def _update_nfs_count(current, new):
-            # type: (cfs.CephFilesystem, cfs.CephFilesystem) -> cfs.CephFilesystem
-            new.spec.metadataServer.activeCount = newcount
-            return new
-        return self._patch(cnfs.CephNFS, 'cephnfses', svc_id, _update_nfs_count)
-
     def update_nfs_count(self, svc_id, newcount):
         def _update_nfs_count(current, new):
             # type: (cnfs.CephNFS, cnfs.CephNFS) -> cnfs.CephNFS
@@ -582,3 +581,44 @@ class RookCluster(object):
                 "Failed to update {}/{}: {}".format(crd_name, cr_name, e))
 
         return "Success"
+
+    def _create_or_patch(self, crd, crd_name, cr_name, update_func, create_func):
+        try:
+            current_json = self.rook_api_get(
+                "{}/{}".format(crd_name, cr_name)
+            )
+        except ApiException as e:
+            if e.status == 404:
+                current_json = None
+            else:
+                raise
+
+        if current_json:
+            current = crd.from_json(current_json)
+            new = crd.from_json(current_json)  # no deepcopy.
+
+            new = update_func(current, new)
+
+            patch = list(jsonpatch.make_patch(current_json, new.to_json()))
+
+            log.info('patch for {}/{}: \n{}'.format(crd_name, cr_name, patch))
+
+            if len(patch) == 0:
+                return "No change"
+
+            try:
+                self.rook_api_patch(
+                    "{}/{}".format(crd_name, cr_name),
+                    body=patch)
+            except ApiException as e:
+                log.exception("API exception: {0}".format(e))
+                raise ApplyException(
+                    "Failed to update {}/{}: {}".format(crd_name, cr_name, e))
+            return "Updated"
+        else:
+            new = create_func()
+            with self.ignore_409("{} {} already exists".format(crd_name,
+                                                               cr_name)):
+                self.rook_api_post("{}/".format(crd_name),
+                                   body=new.to_json())
+            return "Created"