Dev: corosync: implement removing links (jsc#PED-8083)

crmsh/corosync.py

@@ -728,3 +728,38 @@ def update_node_addr(self, linknumber: int, node_addresses: typing.Mapping[int,
             if updated_addr is not None:
                 node[f'ring{linknumber}_addr'] = updated_addr
         return self._config
+
+    def remove_link(self, linknumber: int) -> dict:
+        """Remove the specified link.
+
+        Parameters:
+            * linknumber: the link to update
+        Returns: updated configuration dom. The internal state of LinkManager is also updated.
+        """
+        links = self.links()
+        if linknumber >= len(links):
+            raise ValueError(f'Link {linknumber} does not exist.')
+        if linknumber == 0:
+            raise ValueError(f'Cannot remove the last link.')
+        nodes = self._config['nodelist']['node']
+        assert isinstance(nodes, list)
+        for node in nodes:
+            for key in node:
+                match = re.match('^ring([0-9]+)_addr$', key)
+                if match:
+                    x = int(match.group(1))
+                    if x > linknumber:
+                        node[f'ring{x-1}_addr'] = node[key]
+            del node[f'ring{len(links)-1}_addr']
+        assert 'totem' in self._config
+        if 'interface' not in self._config['totem']:
+            return self._config
+        interfaces = self._config['totem']['interface']
+        assert isinstance(interfaces, list)
+        interfaces = [interface for interface in interfaces if int(interface['linknumber']) != linknumber]
+        for interface in interfaces:
+            ln = int(interface['linknumber'])
+            if ln > linknumber:
+                interface['linknumber'] = str(ln - 1)
+        self._config['totem']['interface'] = interfaces
+        return self._config

crmsh/ui_corosync.py

@@ -166,6 +166,20 @@ def do_update(self, context, *argv):
         logger.info("Use \"crm corosync diff\" to show the difference")
         logger.info("Use \"crm corosync push\" to sync")
 
+    def do_remove(self, context, linknumber: str):
+        if not linknumber.isdecimal():
+            raise ValueError(f'Invalid linknumber: {linknumber}')
+        linknumber = int(linknumber)
+        lm = corosync.LinkManager.load_config_file()
+        if lm.totem_transport() != 'knet':
+            logger.error('Corosync is not using knet transport')
+            return False
+        lm.write_config_file(
+            lm.remove_link(linknumber)
+        )
+        logger.info("Use \"crm corosync diff\" to show the difference")
+        logger.info("Use \"crm corosync push\" to sync")
+
 
 class Corosync(command.UI):
     '''

test/unittests/test_corosync.py

@@ -525,5 +525,71 @@ def test_update_unknown_link(self):
         self.assertDictEqual(self.ORIGINAL, self.lm._config)
 
 
+class TestLinkManagerRemoveLink(unittest.TestCase):
+    ORIGINAL = {
+        'totem': {
+            'interface': [{
+                'linknumber': '0',
+                'knet_link_priority': '1',
+            }, {
+                'linknumber': '2',
+                'knet_link_priority': '10',
+                'knet_transport': 'sctp',
+            }]
+        },
+        'nodelist': {
+            'node': [{
+                'nodeid': '1',
+                'name': 'node1',
+                'ring0_addr': '192.0.2.1',
+                'ring1_addr': '192.0.2.101',
+                'ring2_addr': '192.0.2.201',
+            }, {
+                'nodeid': '3',
+                'name': 'node3',
+                'ring0_addr': '192.0.2.3',
+                'ring1_addr': '192.0.2.103',
+                'ring2_addr': '192.0.2.203',
+            }, {
+                'nodeid': '2',
+                'name': 'node2',
+                'ring0_addr': '192.0.2.3',
+                'ring1_addr': '192.0.2.102',
+                'ring2_addr': '192.0.2.202',
+            }]
+        }
+    }
+
+    def setUp(self):
+        self.lm = corosync.LinkManager(copy.deepcopy(self.ORIGINAL))
+
+    def test_remove(self):
+        self.lm.remove_link(1)
+        self.assertEqual(2, len(self.lm._config['totem']['interface']))
+        self.assertEqual('0', self.lm._config['totem']['interface'][0]['linknumber'])
+        self.assertEqual('1', self.lm._config['totem']['interface'][1]['linknumber'])
+        self.assertEqual(3, len(self.lm._config['nodelist']['node']))
+        self.assertNotIn('ring2_addr', self.lm._config['nodelist']['node'][0])
+        self.assertNotIn('ring2_addr', self.lm._config['nodelist']['node'][1])
+        self.assertNotIn('ring2_addr', self.lm._config['nodelist']['node'][2])
+        self.assertEqual('192.0.2.201', self.lm._config['nodelist']['node'][0]['ring1_addr'])
+        self.assertEqual('192.0.2.203', self.lm._config['nodelist']['node'][1]['ring1_addr'])
+        self.assertEqual('192.0.2.202', self.lm._config['nodelist']['node'][2]['ring1_addr'])
+
+    def test_remove_unknown_link(self):
+        with self.assertRaises(ValueError):
+            self.lm.remove_link(3)
+
+    def test_remove_last_link(self):
+        self.lm.remove_link(1)
+        self.lm.remove_link(1)
+        self.assertEqual(1, len(self.lm._config['totem']['interface']))
+        self.assertNotIn('ring1_addr', self.lm._config['nodelist']['node'][0])
+        self.assertNotIn('ring1_addr', self.lm._config['nodelist']['node'][1])
+        self.assertNotIn('ring1_addr', self.lm._config['nodelist']['node'][2])
+        with self.assertRaises(ValueError):
+            self.lm.remove_link(0)
+
+
 if __name__ == '__main__':
     unittest.main()