Dev: migration: add a checker for cib schema version (jsc#PED-8252)

crmsh/migration.py

@@ -1,5 +1,6 @@
 import argparse
 import dataclasses
+import glob
 import importlib.resources
 import itertools
 import json
@@ -276,7 +277,9 @@ def run(self):
 
 
 def check_global(handler: CheckResultHandler):
-    check_unsupported_resource_agents(handler)
+    cib = xmlutil.text2elem(sh.LocalShell().get_stdout_or_raise_error(None, 'crm configure show xml'))
+    check_cib_schema_version(handler, cib)
+    check_unsupported_resource_agents(handler, cib)
 
 
 def check_dependency_version(handler: CheckResultHandler):
@@ -536,11 +539,10 @@ def nodes(self):
         return result
 
 
-def check_unsupported_resource_agents(handler: CheckResultHandler):
+def check_unsupported_resource_agents(handler: CheckResultHandler, cib: lxml.etree.Element):
     handler.log_info("Checking used resource agents...")
     ocf_resource_agents = list()
     stonith_resource_agents = list()
-    cib = xmlutil.text2elem(sh.LocalShell().get_stdout_or_raise_error(None, 'crm configure show xml'))
     for resource_agent in cibquery.get_configured_resource_agents(cib):
         if resource_agent.m_class == 'ocf':
             ocf_resource_agents.append(resource_agent)
@@ -630,3 +632,36 @@ def _check_ocfs2(handler: CheckResultHandler, cib: lxml.etree.Element):
            True, handler.LEVEL_ERROR,
            'OCFS2 is not supported in SLES 16. Please use GFS2.', [],
        )
+
+def check_cib_schema_version(handler: CheckResultHandler, cib: lxml.etree.Element):
+    schema_version = cib.get('validate-with')
+    if schema_version is None:
+        handler.handle_problem(
+            False, handler.LEVEL_WARN,
+            "The CIB is validated with unknown schema version.", []
+        )
+        return
+    version_match = re.match(r'^pacemaker-(\d+)\.(\d+)$', schema_version)
+    if version_match is None:
+        handler.handle_problem(
+            False, handler.LEVEL_WARN,
+            f"The CIB is validated with unknown schema version {schema_version}", []
+        )
+        return
+    version = tuple(int(x) for x in version_match.groups())
+    latest_schema_version = _get_latest_cib_schema_version()
+    if version != latest_schema_version:
+        handler.handle_problem(
+            False, handler.LEVEL_WARN,
+            "The CIB is not validated with the latest schema version.", [
+                f'* Latest version:  {".".join(str(i) for i in latest_schema_version)}',
+                f'* Current version: {".".join(str(i) for i in version)}',
+            ]
+        )
+
+
+def _get_latest_cib_schema_version() -> tuple[int, int]:
+    return max(tuple(int(s) for s in x.groups()) for x in (
+        re.match(r'^pacemaker-(\d+)\.(\d+)\.rng$', filename)
+        for filename in glob.iglob('pacemaker-*.rng', root_dir='/usr/share/pacemaker')
+    ) if x is not None)

test/unittests/test_migration.py

@@ -2,6 +2,8 @@
 import unittest
 from unittest import mock
 
+import lxml.etree
+
 from crmsh import migration, cibquery
 
 
@@ -56,3 +58,24 @@ def check_fn(x):
         self._handler.handle_problem.assert_not_called()
         check_fn('foo 2.0')
         self._handler.handle_problem.assert_not_called()
+
+    @mock.patch('glob.iglob')
+    def test_get_latest_cib_schema_version(self, mock_iglob: mock.MagicMock):
+        mock_iglob.return_value = iter([
+            'pacemaker-0.1.rng', 'pacemaker-1.9.rng', 'pacemaker-1.11.rng', 'pacemaker-next.rng',
+        ])
+        self.assertEqual((1, 11), migration._get_latest_cib_schema_version())
+
+    @mock.patch('crmsh.migration._get_latest_cib_schema_version')
+    def test_check_cib_schema_version(self, mock_get_latest_cib_schema_version):
+        cib = lxml.etree.fromstring('<cib crm_feature_set="3.16.1" validate-with="pacemaker-3.9" epoch="7" num_updates="0" admin_epoch="0" cib-last-written="Fri Jan  3 13:35:49 2025" update-origin="ha-1-2" update-client="cibadmin" update-user="root" have-quorum="1" dc-uuid="1"/>')
+        mock_get_latest_cib_schema_version.return_value = (3, 10)
+        handler = mock.Mock(migration.CheckResultHandler)
+        migration.check_cib_schema_version(handler, cib)
+        handler.handle_problem.assert_called_with(
+            False, handler.LEVEL_WARN,
+            "The CIB is not validated with the latest schema version.", [
+                '* Latest version:  3.10',
+                '* Current version: 3.9',
+            ]
+        )