mgr/cephadm: unit tests for tuned os profiles

Signed-off-by: Adam King <adking@redhat.com>

diff --git a/src/pybind/mgr/cephadm/tests/test_tuned_profiles.py b/src/pybind/mgr/cephadm/tests/test_tuned_profiles.py
new file mode 100644 (file)
index 0000000..bec433f
--- /dev/null
+++ b/src/pybind/mgr/cephadm/tests/test_tuned_profiles.py
@@ -0,0 +1,216 @@
+import pytest
+import json
+from tests import mock
+from cephadm.tuned_profiles import TunedProfileUtils, SYSCTL_DIR
+from cephadm.inventory import TunedProfileStore
+from ceph.utils import datetime_now
+from ceph.deployment.service_spec import TunedProfileSpec, PlacementSpec
+from cephadm.ssh import SSHManager
+from orchestrator import HostSpec
+
+from typing import List, Dict
+
+
+class SaveError(Exception):
+    pass
+
+
+class FakeCache:
+    def __init__(self,
+                 hosts,
+                 schedulable_hosts,
+                 unreachable_hosts):
+        self.hosts = hosts
+        self.unreachable_hosts = [HostSpec(h) for h in unreachable_hosts]
+        self.schedulable_hosts = [HostSpec(h) for h in schedulable_hosts]
+        self.last_tuned_profile_update = {}
+
+    def get_hosts(self):
+        return self.hosts
+
+    def get_schedulable_hosts(self):
+        return self.schedulable_hosts
+
+    def get_unreachable_hosts(self):
+        return self.unreachable_hosts
+
+    @property
+    def networks(self):
+        return {h: {'a': {'b': ['c']}} for h in self.hosts}
+
+    def host_needs_tuned_profile_update(self, host, profile_name):
+        return profile_name == 'p2'
+
+
+class FakeMgr:
+    def __init__(self,
+                 hosts: List[str],
+                 schedulable_hosts: List[str],
+                 unreachable_hosts: List[str],
+                 profiles: Dict[str, TunedProfileSpec]):
+        self.cache = FakeCache(hosts, schedulable_hosts, unreachable_hosts)
+        self.tuned_profiles = TunedProfileStore(self)
+        self.tuned_profiles.profiles = profiles
+        self.ssh = SSHManager(self)
+
+    def set_store(self, what: str, value: str):
+        raise SaveError(f'{what}: {value}')
+
+    def get_store(self, what: str):
+        if what == 'tuned_profiles':
+            return json.dumps({'x': TunedProfileSpec('x',
+                                                     PlacementSpec(hosts=['x']),
+                                                     {'x': 'x'}).to_json(),
+                               'y': TunedProfileSpec('y',
+                                                     PlacementSpec(hosts=['y']),
+                                                     {'y': 'y'}).to_json()})
+        return ''
+
+
+class TestTunedProfiles:
+    tspec1 = TunedProfileSpec('p1',
+                              PlacementSpec(hosts=['a', 'b', 'c']),
+                              {'setting1': 'value1',
+                               'setting2': 'value2',
+                               'setting with space': 'value with space'})
+    tspec2 = TunedProfileSpec('p2',
+                              PlacementSpec(hosts=['a', 'c']),
+                              {'something': 'something_else',
+                               'high': '5'})
+    tspec3 = TunedProfileSpec('p3',
+                              PlacementSpec(hosts=['c']),
+                              {'wow': 'wow2',
+                               'setting with space': 'value with space',
+                               'down': 'low'})
+
+    def profiles_to_calls(self, tp: TunedProfileUtils, profiles: List[TunedProfileSpec]) -> List[Dict[str, str]]:
+        # this function takes a list of tuned profiles and returns a mapping from
+        # profile names to the string that will be written to the actual config file on the host.
+        res = []
+        for p in profiles:
+            p_str = tp._profile_to_str(p)
+            res.append({p.profile_name: p_str})
+        return res
+
+    @mock.patch("cephadm.tuned_profiles.TunedProfileUtils._remove_stray_tuned_profiles")
+    @mock.patch("cephadm.tuned_profiles.TunedProfileUtils._write_tuned_profiles")
+    def test_write_all_tuned_profiles(self, _write_profiles, _rm_profiles):
+        profiles = {'p1': self.tspec1, 'p2': self.tspec2, 'p3': self.tspec3}
+        mgr = FakeMgr(['a', 'b', 'c'],
+                      ['a', 'b', 'c'],
+                      [],
+                      profiles)
+        tp = TunedProfileUtils(mgr)
+        tp._write_all_tuned_profiles()
+        # need to check that _write_tuned_profiles is correctly called with the
+        # profiles that match the tuned profile placements and with the correct
+        # strings that should be generated from the settings the profiles have.
+        # the _profiles_to_calls helper allows us to generated the input we
+        # should check against
+        calls = [
+            mock.call('a', self.profiles_to_calls(tp, [self.tspec1, self.tspec2])),
+            mock.call('b', self.profiles_to_calls(tp, [self.tspec1])),
+            mock.call('c', self.profiles_to_calls(tp, [self.tspec1, self.tspec2, self.tspec3]))
+        ]
+        _write_profiles.assert_has_calls(calls, any_order=True)
+
+    @mock.patch('cephadm.ssh.SSHManager.check_execute_command')
+    def test_rm_stray_tuned_profiles(self, _check_execute_command):
+        profiles = {'p1': self.tspec1, 'p2': self.tspec2, 'p3': self.tspec3}
+        # for this test, going to use host "a" and put 4 cephadm generated
+        # profiles "p1" "p2", "p3" and "who" only two of which should be there ("p1", "p2")
+        # as well as a file not generated by cephadm. Only the "p3" and "who"
+        # profiles should be removed from the host. This should total to 4
+        # calls to check_execute_command, 1 "ls", 2 "rm", and 1 "sysctl --system"
+        _check_execute_command.return_value = '\n'.join(['p1-cephadm-tuned-profile.conf',
+                                                         'p2-cephadm-tuned-profile.conf',
+                                                         'p3-cephadm-tuned-profile.conf',
+                                                         'who-cephadm-tuned-profile.conf',
+                                                         'dont-touch-me'])
+        mgr = FakeMgr(['a', 'b', 'c'],
+                      ['a', 'b', 'c'],
+                      [],
+                      profiles)
+        tp = TunedProfileUtils(mgr)
+        tp._remove_stray_tuned_profiles('a', self.profiles_to_calls(tp, [self.tspec1, self.tspec2]))
+        calls = [
+            mock.call('a', ['ls', SYSCTL_DIR]),
+            mock.call('a', ['rm', '-f', f'{SYSCTL_DIR}/p3-cephadm-tuned-profile.conf']),
+            mock.call('a', ['rm', '-f', f'{SYSCTL_DIR}/who-cephadm-tuned-profile.conf']),
+            mock.call('a', ['sysctl', '--system'])
+        ]
+        _check_execute_command.assert_has_calls(calls, any_order=True)
+
+    @mock.patch('cephadm.ssh.SSHManager.check_execute_command')
+    @mock.patch('cephadm.ssh.SSHManager.write_remote_file')
+    def test_write_tuned_profiles(self, _write_remote_file, _check_execute_command):
+        profiles = {'p1': self.tspec1, 'p2': self.tspec2, 'p3': self.tspec3}
+        # for this test we will use host "a" and have it so host_needs_tuned_profile_update
+        # returns True for p2 and False for p1 (see FakeCache class). So we should see
+        # 2 ssh calls, one to write p2, one to run sysctl --system
+        _check_execute_command.return_value = 'success'
+        _write_remote_file.return_value = 'success'
+        mgr = FakeMgr(['a', 'b', 'c'],
+                      ['a', 'b', 'c'],
+                      [],
+                      profiles)
+        tp = TunedProfileUtils(mgr)
+        tp._write_tuned_profiles('a', self.profiles_to_calls(tp, [self.tspec1, self.tspec2]))
+        _check_execute_command.assert_called_with('a', ['sysctl', '--system'])
+        _write_remote_file.assert_called_with(
+            'a', f'{SYSCTL_DIR}/p2-cephadm-tuned-profile.conf', tp._profile_to_str(self.tspec2).encode('utf-8'))
+
+    def test_store(self):
+        mgr = FakeMgr(['a', 'b', 'c'],
+                      ['a', 'b', 'c'],
+                      [],
+                      {})
+        tps = TunedProfileStore(mgr)
+        save_str_p1 = 'tuned_profiles: ' + json.dumps({'p1': self.tspec1.to_json()})
+        tspec1_updated = self.tspec1.copy()
+        tspec1_updated.settings.update({'new-setting': 'new-value'})
+        save_str_p1_updated = 'tuned_profiles: ' + json.dumps({'p1': tspec1_updated.to_json()})
+        save_str_p1_updated_p2 = 'tuned_profiles: ' + \
+            json.dumps({'p1': tspec1_updated.to_json(), 'p2': self.tspec2.to_json()})
+        tspec2_updated = self.tspec2.copy()
+        tspec2_updated.settings.pop('something')
+        save_str_p1_updated_p2_updated = 'tuned_profiles: ' + \
+            json.dumps({'p1': tspec1_updated.to_json(), 'p2': tspec2_updated.to_json()})
+        save_str_p2_updated = 'tuned_profiles: ' + json.dumps({'p2': tspec2_updated.to_json()})
+        with pytest.raises(SaveError) as e:
+            tps.add_profile(self.tspec1)
+        assert str(e.value) == save_str_p1
+        assert 'p1' in tps
+        with pytest.raises(SaveError) as e:
+            tps.add_setting('p1', 'new-setting', 'new-value')
+        assert str(e.value) == save_str_p1_updated
+        assert 'new-setting' in tps.list_profiles()[0].settings
+        with pytest.raises(SaveError) as e:
+            tps.add_profile(self.tspec2)
+        assert str(e.value) == save_str_p1_updated_p2
+        assert 'p2' in tps
+        assert 'something' in tps.list_profiles()[1].settings
+        with pytest.raises(SaveError) as e:
+            tps.rm_setting('p2', 'something')
+        assert 'something' not in tps.list_profiles()[1].settings
+        assert str(e.value) == save_str_p1_updated_p2_updated
+        with pytest.raises(SaveError) as e:
+            tps.rm_profile('p1')
+        assert str(e.value) == save_str_p2_updated
+        assert 'p1' not in tps
+        assert 'p2' in tps
+        assert len(tps.list_profiles()) == 1
+        assert tps.list_profiles()[0].profile_name == 'p2'
+
+        cur_last_updated = tps.last_updated('p2')
+        new_last_updated = datetime_now()
+        assert cur_last_updated != new_last_updated
+        tps.set_last_updated('p2', new_last_updated)
+        assert tps.last_updated('p2') == new_last_updated
+
+        # check FakeMgr get_store func to see what is expected to be found in Key Store here
+        tps.load()
+        assert 'x' in tps
+        assert 'y' in tps
+        assert [p for p in tps.list_profiles() if p.profile_name == 'x'][0].settings == {'x': 'x'}
+        assert [p for p in tps.list_profiles() if p.profile_name == 'y'][0].settings == {'y': 'y'}