bootloader: Add rpm-ostree kargs --source=tuned support

jmarrero · jmarrero · commit 263f8d04d381 · 2026-02-26T21:14:16.000-05:00
On bootc systems with transient /etc, the state file
/etc/tuned/bootcmdline is lost after each reboot. This causes
TuneD to lose track of which kernel arguments it previously
set, leading to kargs stacking up on every reboot cycle.

The rpm-ostree --source flag solves this by tracking kargs
ownership directly in the BLS config on /boot, which persists
across reboots regardless of /etc transience. With --source=tuned,
rpm-ostree automatically removes all previous kargs from the
"tuned" source and replaces them with the new set, eliminating
the need for TuneD to maintain its own state for diffing.

This commit adds --source support to the bootloader plugin:

1. Detect --source availability by checking rpm-ostree kargs
   --help output during plugin init.

2. When --source is available, use "rpm-ostree kargs
   --source=tuned --append=X --append=Y" for applying kargs
   and "rpm-ostree kargs --source=tuned" (no --append) for
   clearing all TuneD-owned kargs on profile removal.

3. When --source is not available (older rpm-ostree), fall back
   to the existing --delete/--append + bootcmdline state file
   tracking. No legacy code paths are modified.

4. Continue writing to /etc/tuned/bootcmdline as best-effort
   for diagnostics and backward compatibility, but do not
   depend on it for correctness when --source is available.

5. Add unit tests for the bootloader plugin (the first ever in
   the codebase): 15 tests covering --source detection, apply,
   removal, dispatch logic, error handling, and fallback paths.

Assisted-by: OpenCode (Claude Opus 4.6)
Signed-off-by: Joseph Marrero Corchado &lt;jmarrero@redhat.com&gt;
diff --git a/tests/unit/plugins/test_bootloader.py b/tests/unit/plugins/test_bootloader.py
@@ -0,0 +1,274 @@
+import unittest
+
+try:
+	from unittest.mock import Mock, patch, call
+except ImportError:
+	from mock import Mock, patch, call
+
+import tuned.consts as consts
+
+
+class BootloaderPluginSourceTestCase(unittest.TestCase):
+	"""Tests for the rpm-ostree --source=tuned integration in the bootloader plugin.
+
+	These tests exercise the --source detection, source-based kargs apply,
+	source-based kargs removal, and the dispatch logic that falls back to
+	the legacy code path when --source is not available.
+
+	The BootloaderPlugin constructor requires /etc/grub.d/00_tuned to exist,
+	so we test the methods directly on a minimal mock rather than
+	instantiating the full plugin.
+	"""
+
+	def _make_plugin_mock(self, has_source=True, rpm_ostree_idle=True):
+		"""Create a mock that has the real methods under test patched onto it.
+
+		This avoids instantiating BootloaderPlugin (which requires GRUB2
+		template files on disk) while still testing the actual method logic.
+		"""
+		from tuned.plugins.plugin_bootloader import BootloaderPlugin
+
+		mock_plugin = Mock(spec=BootloaderPlugin)
+		mock_plugin._rpm_ostree_has_source = has_source
+		mock_plugin._cmdline_val = ""
+		mock_plugin._cmd = Mock()
+		mock_plugin._cmd.add_modify_option_in_file = Mock(return_value=True)
+
+		# Make _wait_till_rpm_ostree_idle return the desired value
+		mock_plugin._wait_till_rpm_ostree_idle = Mock(return_value=rpm_ostree_idle)
+
+		# Bind real methods to the mock so we test actual logic
+		mock_plugin._rpm_ostree_has_source_flag = \
+			lambda: BootloaderPlugin._rpm_ostree_has_source_flag(mock_plugin)
+		mock_plugin._rpm_ostree_source_update = \
+			lambda: BootloaderPlugin._rpm_ostree_source_update(mock_plugin)
+		mock_plugin._remove_rpm_ostree_source_tuning = \
+			lambda: BootloaderPlugin._remove_rpm_ostree_source_tuning(mock_plugin)
+		mock_plugin._rpm_ostree_update = \
+			lambda: BootloaderPlugin._rpm_ostree_update(mock_plugin)
+		mock_plugin._remove_rpm_ostree_tuning = \
+			lambda: BootloaderPlugin._remove_rpm_ostree_tuning(mock_plugin)
+		mock_plugin._patch_bootcmdline = \
+			lambda d: BootloaderPlugin._patch_bootcmdline(mock_plugin, d)
+
+		return mock_plugin
+
+	# ------------------------------------------------------------------
+	# _rpm_ostree_has_source_flag() detection
+	# ------------------------------------------------------------------
+
+	def test_has_source_flag_present(self):
+		"""--source flag is detected when present in rpm-ostree kargs --help."""
+		plugin = self._make_plugin_mock()
+		help_text = (
+			"Usage:\n"
+			"  rpm-ostree kargs [OPTION...]\n\n"
+			"  --append=KEY=VALUE     Append kernel argument\n"
+			"  --source=NAME          Track kargs ownership by source name\n"
+			"  --delete=KEY=VALUE     Delete a kernel argument\n"
+		)
+		plugin._cmd.execute = Mock(return_value=(0, help_text, ""))
+		self.assertTrue(plugin._rpm_ostree_has_source_flag())
+		plugin._cmd.execute.assert_called_once_with(
+			['rpm-ostree', 'kargs', '--help'], return_err=True)
+
+	def test_has_source_flag_absent(self):
+		"""--source flag is not detected when absent from rpm-ostree kargs --help."""
+		plugin = self._make_plugin_mock()
+		help_text = (
+			"Usage:\n"
+			"  rpm-ostree kargs [OPTION...]\n\n"
+			"  --append=KEY=VALUE     Append kernel argument\n"
+			"  --delete=KEY=VALUE     Delete a kernel argument\n"
+		)
+		plugin._cmd.execute = Mock(return_value=(0, help_text, ""))
+		self.assertFalse(plugin._rpm_ostree_has_source_flag())
+
+	def test_has_source_flag_in_stderr(self):
+		"""--source flag is detected even if it appears in stderr."""
+		plugin = self._make_plugin_mock()
+		plugin._cmd.execute = Mock(return_value=(0, "", "--source=NAME  Track kargs"))
+		self.assertTrue(plugin._rpm_ostree_has_source_flag())
+
+	def test_has_source_flag_help_fails(self):
+		"""If rpm-ostree kargs --help fails, --source is not detected."""
+		plugin = self._make_plugin_mock()
+		plugin._cmd.execute = Mock(return_value=(1, "", "command not found"))
+		self.assertFalse(plugin._rpm_ostree_has_source_flag())
+
+	# ------------------------------------------------------------------
+	# _rpm_ostree_source_update()
+	# ------------------------------------------------------------------
+
+	def test_source_update_basic(self):
+		"""--source=tuned is called with correct --append flags."""
+		plugin = self._make_plugin_mock()
+		plugin._cmdline_val = "nohz=full isolcpus=1-3"
+		plugin._cmd.execute = Mock(return_value=(0, "", ""))
+
+		plugin._rpm_ostree_source_update()
+
+		plugin._cmd.execute.assert_called_once_with(
+			["rpm-ostree", "kargs", "--source=tuned",
+			 "--append=nohz=full", "--append=isolcpus=1-3"],
+			return_err=True)
+		# Verify bootcmdline state file is updated
+		plugin._cmd.add_modify_option_in_file.assert_called_once_with(
+			consts.BOOT_CMDLINE_FILE,
+			{consts.BOOT_CMDLINE_TUNED_VAR: "nohz=full isolcpus=1-3"})
+
+	def test_source_update_empty_cmdline(self):
+		"""When cmdline is empty, only --source=tuned is passed (clears kargs)."""
+		plugin = self._make_plugin_mock()
+		plugin._cmdline_val = ""
+		plugin._cmd.execute = Mock(return_value=(0, "", ""))
+
+		plugin._rpm_ostree_source_update()
+
+		plugin._cmd.execute.assert_called_once_with(
+			["rpm-ostree", "kargs", "--source=tuned"],
+			return_err=True)
+		plugin._cmd.add_modify_option_in_file.assert_called_once_with(
+			consts.BOOT_CMDLINE_FILE,
+			{consts.BOOT_CMDLINE_TUNED_VAR: ""})
+
+	def test_source_update_rpm_ostree_busy(self):
+		"""When rpm-ostree is busy, source update does not execute."""
+		plugin = self._make_plugin_mock(rpm_ostree_idle=False)
+		plugin._cmdline_val = "nohz=full"
+		plugin._cmd.execute = Mock()
+
+		plugin._rpm_ostree_source_update()
+
+		plugin._cmd.execute.assert_not_called()
+		plugin._cmd.add_modify_option_in_file.assert_not_called()
+
+	def test_source_update_command_fails(self):
+		"""When rpm-ostree kargs --source fails, bootcmdline is not updated."""
+		plugin = self._make_plugin_mock()
+		plugin._cmdline_val = "nohz=full"
+		plugin._cmd.execute = Mock(return_value=(1, "", "error: unknown option"))
+
+		plugin._rpm_ostree_source_update()
+
+		plugin._cmd.execute.assert_called_once()
+		plugin._cmd.add_modify_option_in_file.assert_not_called()
+
+	# ------------------------------------------------------------------
+	# _remove_rpm_ostree_source_tuning()
+	# ------------------------------------------------------------------
+
+	def test_remove_source_tuning(self):
+		"""Removal calls --source=tuned with no --append (clears all)."""
+		plugin = self._make_plugin_mock()
+		plugin._cmd.execute = Mock(return_value=(0, "", ""))
+
+		plugin._remove_rpm_ostree_source_tuning()
+
+		plugin._cmd.execute.assert_called_once_with(
+			["rpm-ostree", "kargs", "--source=tuned"],
+			return_err=True)
+		plugin._cmd.add_modify_option_in_file.assert_called_once_with(
+			consts.BOOT_CMDLINE_FILE,
+			{consts.BOOT_CMDLINE_TUNED_VAR: ""})
+
+	def test_remove_source_tuning_rpm_ostree_busy(self):
+		"""When rpm-ostree is busy, source removal does not execute."""
+		plugin = self._make_plugin_mock(rpm_ostree_idle=False)
+		plugin._cmd.execute = Mock()
+
+		plugin._remove_rpm_ostree_source_tuning()
+
+		plugin._cmd.execute.assert_not_called()
+		# bootcmdline should not be touched either
+		plugin._cmd.add_modify_option_in_file.assert_not_called()
+
+	def test_remove_source_tuning_command_fails(self):
+		"""When --source removal fails, bootcmdline is still cleared (best-effort)."""
+		plugin = self._make_plugin_mock()
+		plugin._cmd.execute = Mock(return_value=(1, "", "error"))
+
+		plugin._remove_rpm_ostree_source_tuning()
+
+		# bootcmdline should still be cleared even on failure
+		plugin._cmd.add_modify_option_in_file.assert_called_once_with(
+			consts.BOOT_CMDLINE_FILE,
+			{consts.BOOT_CMDLINE_TUNED_VAR: ""})
+
+	# ------------------------------------------------------------------
+	# _rpm_ostree_update() dispatch
+	# ------------------------------------------------------------------
+
+	def test_rpm_ostree_update_dispatches_to_source(self):
+		"""When --source is available, _rpm_ostree_update uses the source path."""
+		plugin = self._make_plugin_mock(has_source=True)
+		plugin._cmdline_val = "nohz=full"
+		plugin._cmd.execute = Mock(return_value=(0, "", ""))
+
+		plugin._rpm_ostree_update()
+
+		# Should have called --source=tuned
+		plugin._cmd.execute.assert_called_once_with(
+			["rpm-ostree", "kargs", "--source=tuned", "--append=nohz=full"],
+			return_err=True)
+
+	def test_rpm_ostree_update_fallback_when_no_source(self):
+		"""When --source is not available, _rpm_ostree_update uses the legacy path."""
+		from tuned.plugins.plugin_bootloader import BootloaderPlugin
+
+		plugin = self._make_plugin_mock(has_source=False)
+		plugin._cmdline_val = "nohz=full"
+
+		# Set up legacy path dependencies
+		plugin._get_appended_rpm_ostree_kargs = Mock(return_value=["old_karg=1"])
+		plugin._get_rpm_ostree_kargs = Mock(return_value="root=UUID=xxx old_karg=1")
+		plugin._modify_rpm_ostree_kargs = Mock(return_value=True)
+
+		# Bind _rpm_ostree_update with the real method
+		# but we need to also bind _patch_bootcmdline for the legacy path
+		plugin._rpm_ostree_update = \
+			lambda: BootloaderPlugin._rpm_ostree_update(plugin)
+
+		plugin._rpm_ostree_update()
+
+		# Should NOT have called --source, should use legacy delete/append
+		plugin._modify_rpm_ostree_kargs.assert_called_once_with(
+			delete_kargs=["old_karg=1"], append_kargs=["nohz=full"])
+
+	# ------------------------------------------------------------------
+	# _remove_rpm_ostree_tuning() dispatch
+	# ------------------------------------------------------------------
+
+	def test_remove_rpm_ostree_tuning_dispatches_to_source(self):
+		"""When --source is available, removal uses the source path."""
+		plugin = self._make_plugin_mock(has_source=True)
+		plugin._cmd.execute = Mock(return_value=(0, "", ""))
+
+		plugin._remove_rpm_ostree_tuning()
+
+		plugin._cmd.execute.assert_called_once_with(
+			["rpm-ostree", "kargs", "--source=tuned"],
+			return_err=True)
+
+	def test_remove_rpm_ostree_tuning_fallback_when_no_source(self):
+		"""When --source is not available, removal uses the legacy path."""
+		from tuned.plugins.plugin_bootloader import BootloaderPlugin
+
+		plugin = self._make_plugin_mock(has_source=False)
+		plugin._get_appended_rpm_ostree_kargs = Mock(return_value=["old=1", "old=2"])
+		plugin._modify_rpm_ostree_kargs = Mock(return_value=True)
+
+		plugin._remove_rpm_ostree_tuning = \
+			lambda: BootloaderPlugin._remove_rpm_ostree_tuning(plugin)
+
+		plugin._remove_rpm_ostree_tuning()
+
+		plugin._modify_rpm_ostree_kargs.assert_called_once_with(
+			delete_kargs=["old=1", "old=2"])
+		plugin._cmd.add_modify_option_in_file.assert_called_once_with(
+			consts.BOOT_CMDLINE_FILE,
+			{consts.BOOT_CMDLINE_TUNED_VAR: ""})
+
+
+if __name__ == '__main__':
+	unittest.main()
diff --git a/tuned/plugins/plugin_bootloader.py b/tuned/plugins/plugin_bootloader.py
@@ -193,6 +193,7 @@ def _instance_init(self, instance):
 		self._bls = self._bls_enabled()
 
 		self._rpm_ostree = self._rpm_ostree_status() is not None
+		self._rpm_ostree_has_source = self._rpm_ostree and self._rpm_ostree_has_source_flag()
 
 	def _instance_cleanup(self, instance):
 		pass
@@ -223,6 +224,12 @@ def _rpm_ostree_status(self):
 			return None
 		return splited[1]
 
+	def _rpm_ostree_has_source_flag(self):
+		"""Check if the installed rpm-ostree supports --source for kargs."""
+		(rc, out, err) = self._cmd.execute(
+			['rpm-ostree', 'kargs', '--help'], return_err=True)
+		return '--source=' in (out + err)
+
 	def _wait_till_rpm_ostree_idle(self):
 		"""Check that rpm-ostree is idle, allowing some waiting time."""
 		sleep_cycles = 10
@@ -263,6 +270,45 @@ def _modify_rpm_ostree_kargs(self, delete_kargs=[], append_kargs=[]):
 			return False
 		return True
 
+	def _rpm_ostree_source_update(self):
+		"""Apply kernel parameter tuning using rpm-ostree --source tracking.
+
+		With --source=tuned, rpm-ostree automatically removes all previous
+		kargs from the 'tuned' source and replaces them with the new set.
+		This eliminates the need to track state in /etc/tuned/bootcmdline,
+		which is critical for bootc systems with transient /etc.
+		"""
+		if not self._wait_till_rpm_ostree_idle():
+			log.error("Error modifying rpm-ostree kargs: rpm-ostree is busy")
+			return
+		cmd = ["rpm-ostree", "kargs", "--source=tuned"]
+		for karg in self._cmdline_val.split():
+			cmd.append("--append=%s" % karg)
+		(rc, _, err) = self._cmd.execute(cmd, return_err=True)
+		if rc != 0:
+			log.error("Error applying rpm-ostree kargs with --source: %s" % err)
+			return
+		# Best-effort state file write for diagnostics and backward compat
+		self._patch_bootcmdline({consts.BOOT_CMDLINE_TUNED_VAR: self._cmdline_val})
+
+	def _remove_rpm_ostree_source_tuning(self):
+		"""Remove kernel parameter tuning using rpm-ostree --source tracking.
+
+		Calling --source=tuned with no --append flags clears all kargs
+		owned by the 'tuned' source.
+		"""
+		if not self._wait_till_rpm_ostree_idle():
+			log.error("Error removing rpm-ostree kargs: rpm-ostree is busy")
+			return
+		(rc, _, err) = self._cmd.execute(
+			["rpm-ostree", "kargs", "--source=tuned"], return_err=True)
+		if rc != 0:
+			log.error("Error clearing rpm-ostree kargs source: %s" % err)
+		# Clear the state file even on rpm-ostree failure to prevent stale
+		# state from causing incorrect diffs on the next legacy-path apply.
+		# This matches the behavior of the legacy _remove_rpm_ostree_tuning().
+		self._patch_bootcmdline({consts.BOOT_CMDLINE_TUNED_VAR: ""})
+
 	def _get_effective_options(self, options):
 		"""Merge provided options with plugin default options and merge all cmdline.* options."""
 		effective = self._get_config_options().copy()
@@ -335,6 +381,9 @@ def _get_appended_rpm_ostree_kargs(self):
 
 	def _remove_rpm_ostree_tuning(self):
 		"""Remove kernel parameter tuning in a rpm-ostree system."""
+		if self._rpm_ostree_has_source:
+			self._remove_rpm_ostree_source_tuning()
+			return
 		self._modify_rpm_ostree_kargs(delete_kargs=self._get_appended_rpm_ostree_kargs())
 		self._patch_bootcmdline({consts.BOOT_CMDLINE_TUNED_VAR: ""})
 
@@ -447,6 +496,9 @@ def _grub2_cfg_patch(self, d):
 
 	def _rpm_ostree_update(self):
 		"""Apply kernel parameter tuning in a rpm-ostree system."""
+		if self._rpm_ostree_has_source:
+			self._rpm_ostree_source_update()
+			return
 		appended_kargs = self._get_appended_rpm_ostree_kargs()
 		profile_kargs = self._cmdline_val.split()
 		active_kargs = self._get_rpm_ostree_kargs()
