Dev: pre-migration: check if ocfs2 is used (jsc#PED-11808)

crmsh/cibquery.py

@@ -16,3 +16,10 @@ def get_configured_resource_agents(cib: lxml.etree.Element) -> typing.Set[Resour
         ResourceAgent(e.get('class'), e.get('provider'), e.get('type'))
         for e in cib.xpath('/cib/configuration/resources/primitive')
     )
+
+
+def has_primitive_filesystem_ocfs2(cib: lxml.etree.Element) -> bool:
+    return bool(cib.xpath(
+        '/cib/configuration/resources/primitive[@class="ocf" and @provider="heartbeat" and @type="Filesystem"]'
+        '/instance_attributes/nvpair[@name="fstype" and @value="ocfs2"]'
+    ))

crmsh/migration.py

@@ -7,6 +7,8 @@
 import sys
 import typing
 
+import lxml.etree
+
 from crmsh import cibquery
 from crmsh import constants
 from crmsh import corosync
@@ -296,6 +298,7 @@ def handle_tip(self, title: str, details: typing.Iterable[str]):
         supported_resource_agents,
         stonith_resource_agents,
     )
+    _check_ocfs2(handler, cib)
 
 
 def _check_saphana_resource_agent(handler: CheckResultHandler, resource_agents: typing.Iterable[cibquery.ResourceAgent]):
@@ -317,7 +320,7 @@ def _check_saphana_resource_agent(handler: CheckResultHandler, resource_agents:
             ])
 
 
-def _load_supported_resource_agents() -> typing.Set[cibparser.ResourceAgent]:
+def _load_supported_resource_agents() -> typing.Set[cibquery.ResourceAgent]:
     ret = set()
     for line in pkgutil.get_data(
         'crmsh', 'migration-supported-resource-agents.txt'
@@ -326,18 +329,25 @@ def _load_supported_resource_agents() -> typing.Set[cibparser.ResourceAgent]:
         m_class = parts[0]
         m_provider = parts[1] if len(parts) == 3 else None
         m_type = parts[-1]
-        ret.add(cibparser.ResourceAgent(m_class, m_provider, m_type))
+        ret.add(cibquery.ResourceAgent(m_class, m_provider, m_type))
     return ret
 
 
 
 def _check_removed_resource_agents(
         handler: CheckResultHandler,
-        supported_resource_agents: typing.Set[cibparser.ResourceAgent],
-        resource_agents: typing.Iterable[cibparser.ResourceAgent],
+        supported_resource_agents: typing.Set[cibquery.ResourceAgent],
+        resource_agents: typing.Iterable[cibquery.ResourceAgent],
 ):
     unsupported_resource_agents = [x for x in resource_agents if x not in supported_resource_agents]
     if unsupported_resource_agents:
         handler.handle_problem(False, '', [
             '* ' + ':'.join(x for x in resource_agent if x is not None) for resource_agent in unsupported_resource_agents
         ])
+
+
+def _check_ocfs2(handler: CheckResultHandler, cib: lxml.etree.Element):
+    if cibquery.has_primitive_filesystem_ocfs2(cib):
+       handler.handle_problem(False, 'OCFS2 is not supported in SLES 16.', [
+           '* Before migrating to SLES 16, replace it with GFS2.',
+       ])

test/unittests/test_cibquery.py

@@ -76,3 +76,6 @@ def test_get_resource_agents(self):
             },
             cibquery.get_configured_resource_agents(self.cib),
         )
+
+    def test_has_primitive_filesystem_ocfs2(self):
+        self.assertFalse(cibquery.has_primitive_filesystem_ocfs2(self.cib))