Evaluate the @root volume name also for btrfs

In a volume setup the special volume declaration
<volume name="@root=identifier"/> was only evaluated for the
LVM volume manager. In case of btrfs a hardcoded root volume
name '@' was used. This commit allows to specify a custom
name for the root volume for btrfs as well and also allows
to specify that there should be no such root volume.
Example:

    <volume name="@root=@"/>

Name the root volume '@'. If not specified this stays as
the default to stay compatible

    <volume name="@root=/"/>

Indicate no root volume is wanted. All subvolumes resides
below root (/)

    <volume name="@root=foo"/>

Name the root volume 'foo'

This is related to Issue #2316 and a first patch to
address the requested changes

kiwi/volume_manager/btrfs.py

@@ -74,6 +74,12 @@ def post_init(self, custom_args):
         if 'quota_groups' not in self.custom_args:
             self.custom_args['quota_groups'] = False
 
+        self.root_volume_name = '@'
+        canonical_volume_list = self.get_canonical_volume_list()
+        for volume in canonical_volume_list.volumes:
+            if volume.is_root_volume and volume.name:
+                self.root_volume_name = volume.name
+
         self.subvol_mount_list = []
         self.toplevel_mount = None
         self.toplevel_volume = None
@@ -82,9 +88,9 @@ def setup(self, name=None):
         """
         Setup btrfs volume management
 
-        In case of btrfs a toplevel(@) subvolume is created and marked
+        In case of btrfs an optional toplevel subvolume is created and marked
         as default volume. If snapshots are activated via the custom_args
-        the setup method also created the @/.snapshots/1/snapshot
+        the setup method also creates the .snapshots/1/snapshot
         subvolumes. There is no concept of a volume manager name, thus
         the name argument is not used for btrfs
 
@@ -112,31 +118,38 @@ def setup(self, name=None):
             Command.run(
                 ['btrfs', 'quota', 'enable', self.mountpoint]
             )
-        root_volume = self.mountpoint + '/@'
-        Command.run(
-            ['btrfs', 'subvolume', 'create', root_volume]
-        )
-        if self.custom_args['root_is_snapshot']:
-            snapshot_volume = self.mountpoint + '/@/.snapshots'
+        if self.root_volume_name != '/':
+            root_volume = self.mountpoint + f'/{self.root_volume_name}'
+            Command.run(
+                ['btrfs', 'subvolume', 'create', root_volume]
+            )
+        if self.custom_args['root_is_snapshot'] and \
+           self.root_volume_name != '/':
+            snapshot_volume = self.mountpoint + \
+                f'/{self.root_volume_name}/.snapshots'
             Command.run(
                 ['btrfs', 'subvolume', 'create', snapshot_volume]
             )
             os.chmod(snapshot_volume, 0o700)
             Path.create(snapshot_volume + '/1')
-            snapshot = self.mountpoint + '/@/.snapshots/1/snapshot'
+            snapshot = self.mountpoint + \
+                f'/{self.root_volume_name}/.snapshots/1/snapshot'
             Command.run(
                 ['btrfs', 'subvolume', 'create', snapshot]
             )
-            self._set_default_volume('@/.snapshots/1/snapshot')
-            snapshot = self.mountpoint + '/@/.snapshots/1/snapshot'
-            # Mount /@/.snapshots as /.snapshots inside the root
+            self._set_default_volume(
+                f'{self.root_volume_name}/.snapshots/1/snapshot'
+            )
+            snapshot = self.mountpoint + \
+                f'/{self.root_volume_name}/.snapshots/1/snapshot'
+            # Mount /{some-name}/.snapshots as /.snapshots inside the root
             snapshots_mount = MountManager(
                 device=self.device,
                 mountpoint=snapshot + '/.snapshots'
             )
             self.subvol_mount_list.append(snapshots_mount)
         else:
-            self._set_default_volume('@')
+            self._set_default_volume(self.root_volume_name)
 
     def create_volumes(self, filesystem_name):
         """
@@ -163,12 +176,13 @@ def create_volumes(self, filesystem_name):
 
         for volume in canonical_volume_list.volumes:
             if volume.is_root_volume:
-                # the btrfs root volume named '@' has been created as
+                # the btrfs root volume has been created as
                 # part of the setup procedure
                 pass
             else:
                 log.info('--> sub volume %s', volume.realpath)
-                toplevel = self.mountpoint + '/@/'
+                toplevel = self.mountpoint + \
+                    f'/{self.root_volume_name}/'.replace('//', '')
                 volume_parent_path = os.path.normpath(
                     toplevel + os.path.dirname(volume.realpath)
                 )
@@ -183,15 +197,22 @@ def create_volumes(self, filesystem_name):
                 self.apply_attributes_on_volume(
                     toplevel, volume
                 )
-                if self.custom_args['root_is_snapshot']:
-                    snapshot = self.mountpoint + '/@/.snapshots/1/snapshot/'
-                    volume_mount = MountManager(
-                        device=self.device,
-                        mountpoint=os.path.normpath(snapshot + volume.realpath)
-                    )
-                    self.subvol_mount_list.append(
-                        volume_mount
+                volume_mountpoint = self.mountpoint + \
+                    self.root_volume_name + '/'
+                if self.custom_args['root_is_snapshot'] and \
+                   self.root_volume_name != '/':
+                    volume_mountpoint = self.mountpoint + \
+                        f'/{self.root_volume_name}/.snapshots/1/snapshot/'
+
+                volume_mount = MountManager(
+                    device=self.device,
+                    mountpoint=os.path.normpath(
+                        volume_mountpoint + volume.realpath
                     )
+                )
+                self.subvol_mount_list.append(
+                    volume_mount
+                )
 
     def get_fstab(self, persistency_type='by-label', filesystem_name=None):
         """
@@ -219,7 +240,10 @@ def get_fstab(self, persistency_type='by-label', filesystem_name=None):
             )
             fstab_entry = ' '.join(
                 [
-                    blkid_type + '=' + device_id, subvol_name.replace('@', ''),
+                    blkid_type + '=' + device_id,
+                    subvol_name.replace(
+                        self.root_volume_name, ''
+                    ) if self.root_volume_name != '/' else subvol_name,
                     'btrfs', ','.join(mount_entry_options),
                     '0 {fs_passno}'.format(
                         fs_passno='2' if fs_check else '0'
@@ -245,7 +269,10 @@ def get_volumes(self):
                     'subvol=' + subvol_name
                 ] + self.custom_filesystem_args['mount_options']
             )
-            volumes[subvol_name.replace('@', '')] = {
+            subvol_path = subvol_name.replace(
+                self.root_volume_name, ''
+            ) if self.root_volume_name != '/' else subvol_name
+            volumes[subvol_path] = {
                 'volume_options': subvol_options,
                 'volume_device': volume_mount.device
             }
@@ -309,8 +336,11 @@ def get_mountpoint(self) -> str:
 
         :rtype: string
         """
-        sync_target: List[str] = [self.mountpoint, '@']
-        if self.custom_args.get('root_is_snapshot'):
+        sync_target: List[str] = [self.mountpoint]
+        if self.root_volume_name != '/':
+            sync_target.append(self.root_volume_name)
+        if self.custom_args.get('root_is_snapshot') and \
+           self.root_volume_name != '/':
             sync_target.extend(['.snapshots', '1', 'snapshot'])
         return os.path.join(*sync_target)
 
@@ -325,16 +355,23 @@ def sync_data(self, exclude=None):
         """
         if self.toplevel_mount:
             sync_target = self.get_mountpoint()
-            if self.custom_args['root_is_snapshot']:
+            if self.custom_args['root_is_snapshot'] and \
+               self.root_volume_name != '/':
                 self._create_snapshot_info(
-                    ''.join([self.mountpoint, '/@/.snapshots/1/info.xml'])
+                    ''.join(
+                        [
+                            self.mountpoint,
+                            f'/{self.root_volume_name}/.snapshots/1/info.xml'
+                        ]
+                    )
                 )
             data = DataSync(self.root_dir, sync_target)
             data.sync_data(
                 options=Defaults.get_sync_options(), exclude=exclude
             )
             if self.custom_args['quota_groups'] and \
-               self.custom_args['root_is_snapshot']:
+               self.custom_args['root_is_snapshot'] and \
+               self.root_volume_name != '/':
                 self._create_snapper_quota_configuration()
 
     def set_property_readonly_root(self):
@@ -345,7 +382,8 @@ def set_property_readonly_root(self):
             self.custom_args['root_is_snapshot']
         root_is_readonly_snapshot = \
             self.custom_args['root_is_readonly_snapshot']
-        if root_is_snapshot and root_is_readonly_snapshot:
+        if root_is_snapshot and root_is_readonly_snapshot and \
+           self.root_volume_name != '/':
             sync_target = self.mountpoint
             Command.run(
                 ['btrfs', 'property', 'set', sync_target, 'ro', 'true']
@@ -359,27 +397,28 @@ def _is_volume_enabled_for_fs_check(self, mountpoint):
         return False
 
     def _set_default_volume(self, default_volume):
-        subvolume_list_call = Command.run(
-            ['btrfs', 'subvolume', 'list', self.mountpoint]
-        )
-        for subvolume in subvolume_list_call.output.split('\n'):
-            id_search = re.search('ID (\d+) .*path (.*)', subvolume)
-            if id_search:
-                volume_id = id_search.group(1)
-                volume_path = id_search.group(2)
-                if volume_path == default_volume:
-                    Command.run(
-                        [
-                            'btrfs', 'subvolume', 'set-default',
-                            volume_id, self.mountpoint
-                        ]
-                    )
-                    self.toplevel_volume = default_volume
-                    return
-
-        raise KiwiVolumeRootIDError(
-            'Failed to find btrfs volume: %s' % default_volume
-        )
+        if self.root_volume_name != '/':
+            subvolume_list_call = Command.run(
+                ['btrfs', 'subvolume', 'list', self.mountpoint]
+            )
+            for subvolume in subvolume_list_call.output.split('\n'):
+                id_search = re.search('ID (\d+) .*path (.*)', subvolume)
+                if id_search:
+                    volume_id = id_search.group(1)
+                    volume_path = id_search.group(2)
+                    if volume_path == default_volume:
+                        Command.run(
+                            [
+                                'btrfs', 'subvolume', 'set-default',
+                                volume_id, self.mountpoint
+                            ]
+                        )
+                        self.toplevel_volume = default_volume
+                        return
+
+            raise KiwiVolumeRootIDError(
+                'Failed to find btrfs volume: %s' % default_volume
+            )
 
     def _xml_pretty(self, toplevel_element):
         xml_data_unformatted = ElementTree.tostring(
@@ -389,7 +428,14 @@ def _xml_pretty(self, toplevel_element):
         return xml_data_domtree.toprettyxml(indent="    ")
 
     def _create_snapper_quota_configuration(self):
-        root_path = os.sep.join([self.mountpoint, '@/.snapshots/1/snapshot'])
+        root_path = os.sep.join(
+            [
+                self.mountpoint,
+                f'{self.root_volume_name}/.snapshots/1/snapshot'.replace(
+                    '//', '/'
+                )
+            ]
+        )
         snapper_default_conf = Defaults.get_snapper_config_template_file(
             root_path
         )
@@ -453,7 +499,9 @@ def _get_subvol_name_from_mountpoint(self, volume_mount):
         )
         if self.toplevel_volume and self.toplevel_volume in subvol_name:
             subvol_name = subvol_name.replace(self.toplevel_volume, '')
-        return os.path.normpath(os.sep.join(['@', subvol_name]))
+        return os.path.normpath(
+            os.sep.join([self.root_volume_name, subvol_name]).replace('//', '/')
+        )
 
     def __del__(self):
         if self.toplevel_mount:

kiwi/xml_state.py

@@ -1680,6 +1680,9 @@ def get_volumes(self) -> List[volume_type]:
         if not have_root_volume_setup:
             # There must always be a root volume setup. It will be the
             # full size volume if no other volume has this setup
+            volume_management = self.get_volume_management()
+            root_volume_name = \
+                defaults.ROOT_VOLUME_NAME if volume_management == 'lvm' else ''
             if have_full_size_volume:
                 size = 'freespace:' + format(
                     Defaults.get_min_volume_mbytes()
@@ -1690,7 +1693,7 @@ def get_volumes(self) -> List[volume_type]:
                 fullsize = True
             volume_type_list.append(
                 volume_type(
-                    name=defaults.ROOT_VOLUME_NAME,
+                    name=root_volume_name,
                     size=size,
                     fullsize=fullsize,
                     mountpoint=None,

test/unit/volume_manager/btrfs_test.py

@@ -27,12 +27,12 @@ def inject_fixtures(self, caplog):
     def setup(self, mock_path):
         self.volumes = [
             volume_type(
-                name='LVRoot', size='freespace:100', realpath='/',
+                name='@', size='freespace:100', realpath='/',
                 mountpoint=None, fullsize=False, label=None,
                 attributes=[], is_root_volume=True
             ),
             volume_type(
-                name='LVetc', size='freespace:200', realpath='/etc',
+                name='etc', size='freespace:200', realpath='/etc',
                 mountpoint='/etc', fullsize=False, label=None,
                 attributes=[], is_root_volume=False
             ),
@@ -42,7 +42,7 @@ def setup(self, mock_path):
                 attributes=[], is_root_volume=False
             ),
             volume_type(
-                name='LVhome', size=None, realpath='/home',
+                name='home', size=None, realpath='/home',
                 mountpoint='/home', fullsize=True, label=None,
                 attributes=[], is_root_volume=False
             )
@@ -159,6 +159,47 @@ def test_setup_volume_id_not_detected(
         with raises(KiwiVolumeRootIDError):
             self.volume_manager.setup()
 
+    @patch('os.path.exists')
+    @patch('kiwi.volume_manager.btrfs.Command.run')
+    @patch('kiwi.volume_manager.btrfs.MountManager')
+    @patch('kiwi.volume_manager.btrfs.Path.create')
+    @patch('kiwi.volume_manager.base.VolumeManagerBase.apply_attributes_on_volume')
+    def test_create_volumes_no_toplevel_volume(
+        self, mock_attrs, mock_path, mock_mount, mock_command, mock_os_exists
+    ):
+        volume_mount = Mock()
+        mock_mount.return_value = volume_mount
+        self.volume_manager.mountpoint = 'tmpdir'
+        self.volume_manager.custom_args['root_is_snapshot'] = False
+        mock_os_exists.return_value = False
+
+        self.volume_manager.root_volume_name = '/'
+        self.volume_manager.volumes = [
+            volume_type(
+                name='/', size='freespace:100', realpath='/',
+                mountpoint=None, fullsize=False, label=None,
+                attributes=[], is_root_volume=True
+            ),
+            volume_type(
+                name='home', size=None, realpath='/home',
+                mountpoint='/home', fullsize=True, label=None,
+                attributes=[], is_root_volume=False
+            )
+        ]
+
+        self.volume_manager.create_volumes('btrfs')
+
+        assert mock_path.call_args_list == [
+            call('root_dir/home'),
+            call('tmpdir')
+        ]
+        mock_command.assert_called_once_with(
+            ['btrfs', 'subvolume', 'create', 'tmpdir/home']
+        )
+        mock_mount.assert_called_once_with(
+            device='/dev/storage', mountpoint='tmpdir/home'
+        )
+
     @patch('os.path.exists')
     @patch('kiwi.volume_manager.btrfs.Command.run')
     @patch('kiwi.volume_manager.btrfs.MountManager')
@@ -186,15 +227,15 @@ def test_create_volumes(
             ),
             call(
                 'tmpdir/@/', volume_type(
-                    name='LVetc', size='freespace:200', realpath='/etc',
+                    name='etc', size='freespace:200', realpath='/etc',
                     mountpoint='/etc', fullsize=False, label=None,
                     attributes=[],
                     is_root_volume=False
                 )
             ),
             call(
                 'tmpdir/@/', volume_type(
-                    name='LVhome', size=None, realpath='/home',
+                    name='home', size=None, realpath='/home',
                     mountpoint='/home', fullsize=True, label=None,
                     attributes=[],
                     is_root_volume=False