qa: Added tests for disabling stretch mode

Test disabling stretch mode with the following scenario:

1. Healthy Stretch Mode
2. Degraded Stretch Mode

Fixes: https://tracker.ceph.com/issues/67467

Signed-off-by: Kamoltat Sirivadhna <[email protected]>

Author: kamoltat

qa/suites/rados/singleton/all/stretch-mode-5-mons-8-osds.yaml

@@ -0,0 +1,58 @@
+roles:
+- - mon.a
+  - mon.b
+  - mgr.a
+  - mgr.b
+  - osd.0
+  - osd.1
+  - osd.2
+  - osd.3
+- - mon.c
+  - mon.d
+  - mgr.c
+  - mgr.d
+  - osd.4
+  - osd.5
+  - osd.6
+  - osd.7
+- - mon.e
+- - client.0
+
+openstack:
+  - volumes: # attached to each instance
+      count: 3
+      size: 10 # GB
+overrides:
+  ceph:
+    conf:
+      global:
+        mon election default strategy: 3
+        osd pool default size: 3
+        osd pool default min size: 2
+      mon:
+        debug mon: 30
+tasks:
+- install:
+- ceph:
+    pre-mgr-commands:
+      - sudo ceph config set mgr mgr_pool false --force
+    log-ignorelist:
+      - \(POOL_
+      - \(CACHE_POOL_
+      - overall HEALTH_
+      - \(PG_AVAILABILITY\)
+      - Reduced data availability
+      - \(PG_DEGRADED\)
+      - \(MON_DOWN\)
+      - \(OSD_DATACENTER_DOWN\)
+      - \(OSD_DOWN\)
+      - \(OSD_HOST_DOWN\)
+
+
+- workunit:
+    clients:
+      client.0:
+        - mon/mon-stretch-mode-5-mons-8-osds.sh
+- cephfs_test_runner:
+    modules:
+      - tasks.stretch_mode_disable_enable

qa/tasks/ceph_manager.py

@@ -2796,6 +2796,59 @@ def _get_num_peered(self, pgs):
                  num += 1
         return num
 
+    def _print_not_active_clean_pg(self, pgs):
+        """
+        Print the PGs that are not active+clean.
+        """
+        for pg in pgs:
+            if not (pg['state'].count('active') and
+                    pg['state'].count('clean') and
+                    not pg['state'].count('stale')):
+                log.debug(
+                    "PG %s is not active+clean, but %s",
+                    pg['pgid'], pg['state']
+                )
+
+    def pg_all_active_clean(self):
+        """
+        Check if all pgs are active+clean
+        return: True if all pgs are active+clean else False
+        """
+        pgs = self.get_pg_stats()
+        result = self._get_num_active_clean(pgs) == len(pgs)
+        if result:
+            log.debug("All PGs are active+clean")
+        else:
+            log.debug("Not all PGs are active+clean")
+            self._print_not_active_clean_pg(pgs)
+        return result
+
+    def _print_not_active_pg(self, pgs):
+        """
+        Print the PGs that are not active.
+        """
+        for pg in pgs:
+            if not (pg['state'].count('active')
+                    and not pg['state'].count('stale')):
+                log.debug(
+                    "PG %s is not active, but %s",
+                    pg['pgid'], pg['state']
+                )
+
+    def pg_all_active(self):
+        """
+        Check if all pgs are active
+        return: True if all pgs are active else False
+        """
+        pgs = self.get_pg_stats()
+        result = self._get_num_active(pgs) == len(pgs)
+        if result:
+            log.debug("All PGs are active")
+        else:
+            log.debug("Not all PGs are active")
+            self._print_not_active_pg(pgs)
+        return result
+
     def is_clean(self):
         """
         True if all pgs are clean
@@ -3237,6 +3290,26 @@ def revive_mgr(self, mgr):
             self.make_admin_daemon_dir(remote)
         self.ctx.daemons.get_daemon('mgr', mgr, self.cluster).restart()
 
+    def get_crush_rule_id(self, crush_rule_name):
+        """
+        Get crush rule id by name
+        :returns: int -- crush rule id
+        """
+        out = self.raw_cluster_cmd('osd', 'crush', 'rule', 'dump', '--format=json')
+        j = json.loads('\n'.join(out.split('\n')[1:]))
+        for rule in j:
+            if rule['rule_name'] == crush_rule_name:
+                return rule['rule_id']
+        assert False, 'rule %s not found' % crush_rule_name
+
+    def get_mon_dump_json(self):
+        """
+        mon dump --format=json converted to a python object
+        :returns: the python object
+        """
+        out = self.raw_cluster_cmd('mon', 'dump', '--format=json')
+        return json.loads('\n'.join(out.split('\n')[1:]))
+
     def get_mon_status(self, mon):
         """
         Extract all the monitor status information from the cluster
@@ -3340,6 +3413,23 @@ def get_service_task_status(self, service, status_key):
         self.log(task_status)
         return task_status
 
+    # Stretch mode related functions
+    def is_degraded_stretch_mode(self):
+        """
+        Return whether the cluster is in degraded stretch mode
+        """
+        try:
+            osdmap = self.get_osd_dump_json()
+            stretch_mode = osdmap.get('stretch_mode', {})
+            degraded_stretch_mode = stretch_mode.get('degraded_stretch_mode', 0)
+            self.log("is_degraded_stretch_mode: {0}".format(degraded_stretch_mode))
+            return degraded_stretch_mode == 1
+        except (TypeError, AttributeError) as e:
+            # Log the error or handle it as needed
+            self.log("Error accessing degraded_stretch_mode: {0}".format(e))
+            return False
+
+
 def utility_task(name):
     """
     Generate ceph_manager subtask corresponding to ceph_manager