Refactor config drive metadata service

Refactor the existing ConfigDrive Metadata service so that another
metadata format implemenation like NoCloud can be easily added.

Now, the drive label and the metadata file can be set in the
constructor, making it easy to add another config drive metadata service
with different label or path.

Change-Id: I8dd8160dfbe9f529bb8f30ab85181f264c18833e

Author: ader1990

cloudbaseinit/metadata/services/baseconfigdrive.py

@@ -0,0 +1,93 @@
+# Copyright 2012 Cloudbase Solutions Srl
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import os
+import shutil
+
+from oslo_log import log as oslo_logging
+
+from cloudbaseinit import conf as cloudbaseinit_conf
+from cloudbaseinit import constant
+from cloudbaseinit import exception
+from cloudbaseinit.metadata.services import base
+from cloudbaseinit.metadata.services.osconfigdrive import factory
+
+CONF = cloudbaseinit_conf.CONF
+LOG = oslo_logging.getLogger(__name__)
+
+CD_TYPES = constant.CD_TYPES
+CD_LOCATIONS = constant.CD_LOCATIONS
+
+
+class BaseConfigDriveService(base.BaseMetadataService):
+
+    def __init__(self, drive_label, metadata_file):
+        super(BaseConfigDriveService, self).__init__()
+        self._drive_label = drive_label
+        self._metadata_file = metadata_file
+        self._metadata_path = None
+        self._searched_types = set()
+        self._searched_locations = set()
+
+    def _preprocess_options(self):
+        self._searched_types = set(CONF.config_drive.types)
+        self._searched_locations = set(CONF.config_drive.locations)
+
+        # Deprecation backward compatibility.
+        if CONF.config_drive.raw_hdd:
+            self._searched_types.add("iso")
+            self._searched_locations.add("hdd")
+        if CONF.config_drive.cdrom:
+            self._searched_types.add("iso")
+            self._searched_locations.add("cdrom")
+        if CONF.config_drive.vfat:
+            self._searched_types.add("vfat")
+            self._searched_locations.add("hdd")
+
+        # Check for invalid option values.
+        if self._searched_types | CD_TYPES != CD_TYPES:
+            raise exception.CloudbaseInitException(
+                "Invalid Config Drive types %s", self._searched_types)
+        if self._searched_locations | CD_LOCATIONS != CD_LOCATIONS:
+            raise exception.CloudbaseInitException(
+                "Invalid Config Drive locations %s", self._searched_locations)
+
+    def load(self):
+        super(BaseConfigDriveService, self).load()
+
+        self._preprocess_options()
+        self._mgr = factory.get_config_drive_manager()
+        found = self._mgr.get_config_drive_files(
+            drive_label=self._drive_label,
+            metadata_file=self._metadata_file,
+            searched_types=self._searched_types,
+            searched_locations=self._searched_locations)
+
+        if found:
+            self._metadata_path = self._mgr.target_path
+            LOG.debug('Metadata copied to folder: %r', self._metadata_path)
+        return found
+
+    def _get_data(self, path):
+        norm_path = os.path.normpath(os.path.join(self._metadata_path, path))
+        try:
+            with open(norm_path, 'rb') as stream:
+                return stream.read()
+        except IOError:
+            raise base.NotExistingMetadataException()
+
+    def cleanup(self):
+        LOG.debug('Deleting metadata folder: %r', self._mgr.target_path)
+        shutil.rmtree(self._mgr.target_path, ignore_errors=True)
+        self._metadata_path = None

cloudbaseinit/metadata/services/configdrive.py

@@ -1,4 +1,4 @@
-# Copyright 2012 Cloudbase Solutions Srl
+# Copyright 2020 Cloudbase Solutions Srl
 #
 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
 #    not use this file except in compliance with the License. You may obtain
@@ -12,77 +12,19 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import os
-import shutil
-
 from oslo_log import log as oslo_logging
 
 from cloudbaseinit import conf as cloudbaseinit_conf
-from cloudbaseinit import constant
-from cloudbaseinit import exception
-from cloudbaseinit.metadata.services import base
+from cloudbaseinit.metadata.services import baseconfigdrive
 from cloudbaseinit.metadata.services import baseopenstackservice
-from cloudbaseinit.metadata.services.osconfigdrive import factory
 
 CONF = cloudbaseinit_conf.CONF
 LOG = oslo_logging.getLogger(__name__)
 
-CD_TYPES = constant.CD_TYPES
-CD_LOCATIONS = constant.CD_LOCATIONS
-
 
-class ConfigDriveService(baseopenstackservice.BaseOpenStackService):
+class ConfigDriveService(baseconfigdrive.BaseConfigDriveService,
+                         baseopenstackservice.BaseOpenStackService):
 
     def __init__(self):
-        super(ConfigDriveService, self).__init__()
-        self._metadata_path = None
-
-    def _preprocess_options(self):
-        self._searched_types = set(CONF.config_drive.types)
-        self._searched_locations = set(CONF.config_drive.locations)
-
-        # Deprecation backward compatibility.
-        if CONF.config_drive.raw_hdd:
-            self._searched_types.add("iso")
-            self._searched_locations.add("hdd")
-        if CONF.config_drive.cdrom:
-            self._searched_types.add("iso")
-            self._searched_locations.add("cdrom")
-        if CONF.config_drive.vfat:
-            self._searched_types.add("vfat")
-            self._searched_locations.add("hdd")
-
-        # Check for invalid option values.
-        if self._searched_types | CD_TYPES != CD_TYPES:
-            raise exception.CloudbaseInitException(
-                "Invalid Config Drive types %s", self._searched_types)
-        if self._searched_locations | CD_LOCATIONS != CD_LOCATIONS:
-            raise exception.CloudbaseInitException(
-                "Invalid Config Drive locations %s", self._searched_locations)
-
-    def load(self):
-        super(ConfigDriveService, self).load()
-
-        self._preprocess_options()
-        self._mgr = factory.get_config_drive_manager()
-        found = self._mgr.get_config_drive_files(
-            searched_types=self._searched_types,
-            searched_locations=self._searched_locations)
-
-        if found:
-            self._metadata_path = self._mgr.target_path
-            LOG.debug('Metadata copied to folder: %r', self._metadata_path)
-        return found
-
-    def _get_data(self, path):
-        norm_path = os.path.normpath(os.path.join(self._metadata_path, path))
-        try:
-            with open(norm_path, 'rb') as stream:
-                return stream.read()
-        except IOError:
-            raise base.NotExistingMetadataException()
-
-    def cleanup(self):
-        LOG.debug('Deleting metadata folder: %r', self._mgr.target_path)
-        shutil.rmtree(self._mgr.target_path, ignore_errors=True)
-        self._metadata_path = None
+        super(ConfigDriveService, self).__init__(
+            'config-2', 'openstack\\latest\\meta_data.json')

cloudbaseinit/metadata/services/osconfigdrive/base.py

@@ -25,5 +25,7 @@ def __init__(self):
         self.target_path = tempfile.mkdtemp()
 
     @abc.abstractmethod
-    def get_config_drive_files(self, check_types=None, check_locations=None):
+    def get_config_drive_files(self, drive_label, metadata_file,
+                               check_types=None, check_locations=None,
+                               ):
         pass

cloudbaseinit/metadata/services/osconfigdrive/windows.py

@@ -32,7 +32,6 @@
 CONF = cloudbaseinit_conf.CONF
 LOG = oslo_logging.getLogger(__name__)
 
-CONFIG_DRIVE_LABEL = 'config-2'
 MAX_SECTOR_SIZE = 4096
 # Absolute offset values and the ISO magic string.
 OFFSET_BOOT_RECORD = 0x8000
@@ -50,18 +49,25 @@ def __init__(self):
         super(WindowsConfigDriveManager, self).__init__()
         self._osutils = osutils_factory.get_os_utils()
 
-    def _check_for_config_drive(self, drive):
+    def _meta_data_file_exists(self, drive, metadata_file):
+        metadata_file = os.path.join(drive, metadata_file)
+
+        if os.path.exists(metadata_file):
+            return True
+
+        LOG.debug('%s not found', metadata_file)
+        return False
+
+    def _check_for_config_drive(self, drive, required_drive_label,
+                                metadata_file):
         label = self._osutils.get_volume_label(drive)
-        if label and label.lower() == CONFIG_DRIVE_LABEL:
-            if os.path.exists(os.path.join(drive,
-                              'openstack\\latest\\meta_data.json')):
-                LOG.info('Config Drive found on %s', drive)
-                return True
-            LOG.debug('%s\openstack\latest\meta_data.json not '
-                      'found', drive)
+        if label and label.lower() == required_drive_label and \
+                self._meta_data_file_exists(drive, metadata_file):
+            LOG.info('Config Drive found on %s', drive)
+            return True
         LOG.debug("Looking for a Config Drive with label '%s' on '%s'. "
                   "Found mismatching label '%s'.",
-                  CONFIG_DRIVE_LABEL, drive, label)
+                  required_drive_label, drive, label)
         return False
 
     def _get_iso_file_size(self, device):
@@ -137,20 +143,21 @@ def _extract_iso_from_devices(self, devices):
             os.remove(iso_file_path)
         return extracted
 
-    def _get_config_drive_from_cdrom_drive(self):
+    def _get_config_drive_from_cdrom_drive(self, drive_label, metadata_file):
         for drive_letter in self._osutils.get_cdrom_drives():
-            if self._check_for_config_drive(drive_letter):
+            if self._check_for_config_drive(drive_letter, drive_label,
+                                            metadata_file):
                 os.rmdir(self.target_path)
                 shutil.copytree(drive_letter, self.target_path)
                 return True
 
         return False
 
-    def _get_config_drive_from_raw_hdd(self):
+    def _get_config_drive_from_raw_hdd(self, drive_label, metadata_file):
         disks = map(disk.Disk, self._osutils.get_physical_disks())
         return self._extract_iso_from_devices(disks)
 
-    def _get_config_drive_from_vfat(self):
+    def _get_config_drive_from_vfat(self, drive_label, metadata_file):
         for drive_path in self._osutils.get_physical_disks():
             if vfat.is_vfat_drive(self._osutils, drive_path):
                 LOG.info('Config Drive found on disk %r', drive_path)
@@ -159,7 +166,7 @@ def _get_config_drive_from_vfat(self):
                 return True
         return False
 
-    def _get_config_drive_from_partition(self):
+    def _get_config_drive_from_partition(self, drive_label, metadata_file):
         for disk_path in self._osutils.get_physical_disks():
             physical_drive = disk.Disk(disk_path)
             with physical_drive:
@@ -169,22 +176,24 @@ def _get_config_drive_from_partition(self):
                 return True
         return False
 
-    def _get_config_drive_from_volume(self):
+    def _get_config_drive_from_volume(self, drive_label, metadata_file):
         """Look through all the volumes for config drive."""
         volumes = self._osutils.get_volumes()
         for volume in volumes:
-            if self._check_for_config_drive(volume):
+            if self._check_for_config_drive(volume, drive_label,
+                                            metadata_file):
                 os.rmdir(self.target_path)
                 shutil.copytree(volume, self.target_path)
                 return True
         return False
 
-    def _get_config_drive_files(self, cd_type, cd_location):
+    def _get_config_drive_files(self, drive_label, metadata_file,
+                                cd_type, cd_location):
         try:
             get_config_drive = self.config_drive_type_location.get(
                 "{}_{}".format(cd_location, cd_type))
             if get_config_drive:
-                return get_config_drive()
+                return get_config_drive(drive_label, metadata_file)
             else:
                 LOG.debug("Irrelevant type %(type)s in %(location)s "
                           "location; skip",
@@ -196,16 +205,19 @@ def _get_config_drive_files(self, cd_type, cd_location):
 
         return False
 
-    def get_config_drive_files(self, searched_types=None,
-                               searched_locations=None):
+    def get_config_drive_files(self, drive_label, metadata_file,
+                               searched_types=None, searched_locations=None):
         searched_types = searched_types or []
         searched_locations = searched_locations or []
 
         for cd_type, cd_location in itertools.product(searched_types,
                                                       searched_locations):
-            LOG.debug('Looking for Config Drive %(type)s in %(location)s',
-                      {"type": cd_type, "location": cd_location})
-            if self._get_config_drive_files(cd_type, cd_location):
+            LOG.debug('Looking for Config Drive %(type)s in %(location)s '
+                      'with expected label %(drive_label)s',
+                      {"type": cd_type, "location": cd_location,
+                       "drive_label": drive_label})
+            if self._get_config_drive_files(drive_label, metadata_file,
+                                            cd_type, cd_location):
                 return True
 
         return False