openstack/ironic
Code Issues Proposed changes

Reduce the number of small functions in pxe_utils

Browse Source 
This module contains a lot of trivial single-line functions that make
reading it unnecessarily complicated. This change removes some of them.

Change-Id: Ie26c011b5ed3cb406c5cc4e2f50bbbf622d3a8c8
This commit is contained in:
Dmitry Tantsur 2021-11-15 18:21:13 +01:00
parent 2ef65aa368
commit 545a222a04
2 changed files with 35 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

103
ironic/common/pxe_utils.py
View File

@ -65,13 +65,11 @@ KERNEL_RAMDISK_LABELS = {'deploy': DEPLOY_KERNEL_RAMDISK_LABELS,
'rescue': RESCUE_KERNEL_RAMDISK_LABELS} 'rescue': RESCUE_KERNEL_RAMDISK_LABELS}
def get_root_dir(): def _get_root_dir(ipxe_enabled):
"""Returns the directory where the config files and images will live.""" if ipxe_enabled:
return CONF.pxe.tftp_root return CONF.deploy.http_root
else:
return CONF.pxe.tftp_root
def get_ipxe_root_dir():
return CONF.deploy.http_root
def _ensure_config_dirs_exist(task, ipxe_enabled=False): def _ensure_config_dirs_exist(task, ipxe_enabled=False):
@ -81,10 +79,7 @@ def _ensure_config_dirs_exist(task, ipxe_enabled=False):
:param ipxe_enabled: Default false boolean to indicate if ipxe :param ipxe_enabled: Default false boolean to indicate if ipxe
is in use by the caller. is in use by the caller.
""" """
if ipxe_enabled: root_dir = _get_root_dir(ipxe_enabled)
root_dir = get_ipxe_root_dir()
else:
root_dir = get_root_dir()
node_dir = os.path.join(root_dir, task.node.uuid) node_dir = os.path.join(root_dir, task.node.uuid)
pxe_dir = os.path.join(root_dir, PXE_CFG_DIR_NAME) pxe_dir = os.path.join(root_dir, PXE_CFG_DIR_NAME)
# NOTE: We should only change the permissions if the folder # NOTE: We should only change the permissions if the folder
@ -164,7 +159,7 @@ def _link_ip_address_pxe_configs(task, ipxe_enabled=False):
def _get_pxe_grub_mac_path(mac, ipxe_enabled=False): def _get_pxe_grub_mac_path(mac, ipxe_enabled=False):
root_dir = get_ipxe_root_dir() if ipxe_enabled else get_root_dir() root_dir = _get_root_dir(ipxe_enabled)
yield os.path.join(root_dir, "%s-%s-%s" % yield os.path.join(root_dir, "%s-%s-%s" %
("grub.cfg", "01", mac.replace(':', "-").lower())) ("grub.cfg", "01", mac.replace(':', "-").lower()))
yield os.path.join(root_dir, mac + '.conf') yield os.path.join(root_dir, mac + '.conf')
@ -189,9 +184,9 @@ def _get_pxe_mac_path(mac, delimiter='-', client_id=None,
if client_id: if client_id:
hw_type = '20-' hw_type = '20-'
mac_file_name = hw_type + mac_file_name mac_file_name = hw_type + mac_file_name
return os.path.join(get_root_dir(), PXE_CFG_DIR_NAME, return os.path.join(CONF.pxe.tftp_root, PXE_CFG_DIR_NAME,
mac_file_name) mac_file_name)
return os.path.join(get_ipxe_root_dir(), PXE_CFG_DIR_NAME, return os.path.join(CONF.deploy.http_root, PXE_CFG_DIR_NAME,
mac_file_name) mac_file_name)
@ -226,10 +221,7 @@ def get_kernel_ramdisk_info(node_uuid, driver_info, mode='deploy',
Note: driver_info should be validated outside of this method. Note: driver_info should be validated outside of this method.
""" """
if ipxe_enabled: root_dir = _get_root_dir(ipxe_enabled)
root_dir = get_ipxe_root_dir()
else:
root_dir = get_root_dir()
image_info = {} image_info = {}
labels = KERNEL_RAMDISK_LABELS[mode] labels = KERNEL_RAMDISK_LABELS[mode]
for label in labels: for label in labels:
@ -249,10 +241,7 @@ def get_pxe_config_file_path(node_uuid, ipxe_enabled=False):
:returns: The path to the node's PXE configuration file. :returns: The path to the node's PXE configuration file.
""" """
if ipxe_enabled: return os.path.join(_get_root_dir(ipxe_enabled), node_uuid, 'config')
return os.path.join(get_ipxe_root_dir(), node_uuid, 'config')
else:
return os.path.join(get_root_dir(), node_uuid, 'config')
def get_file_path_from_label(node_uuid, root_dir, label): def get_file_path_from_label(node_uuid, root_dir, label):
@ -269,11 +258,12 @@ def get_file_path_from_label(node_uuid, root_dir, label):
:param label: Name of the image :param label: Name of the image
""" """
if label == 'ks_template': if label == 'ks_template':
return os.path.join(get_ipxe_root_dir(), node_uuid, 'ks.cfg.template') return os.path.join(CONF.deploy.http_root, node_uuid,
'ks.cfg.template')
elif label == 'ks_cfg': elif label == 'ks_cfg':
return os.path.join(get_ipxe_root_dir(), node_uuid, 'ks.cfg') return os.path.join(CONF.deploy.http_root, node_uuid, 'ks.cfg')
elif label == 'stage2': elif label == 'stage2':
return os.path.join(get_ipxe_root_dir(), node_uuid, 'LiveOS', return os.path.join(CONF.deploy.http_root, node_uuid, 'LiveOS',
'squashfs.img') 'squashfs.img')
else: else:
return os.path.join(root_dir, node_uuid, label) return os.path.join(root_dir, node_uuid, label)
@ -426,12 +416,8 @@ def clean_up_pxe_config(task, ipxe_enabled=False):
for path in _get_pxe_grub_mac_path(port.address, for path in _get_pxe_grub_mac_path(port.address,
ipxe_enabled=ipxe_enabled): ipxe_enabled=ipxe_enabled):
ironic_utils.unlink_without_raise(path) ironic_utils.unlink_without_raise(path)
if ipxe_enabled: utils.rmtree_without_raise(os.path.join(_get_root_dir(ipxe_enabled),
utils.rmtree_without_raise(os.path.join(get_ipxe_root_dir(), task.node.uuid))
task.node.uuid))
else:
utils.rmtree_without_raise(os.path.join(get_root_dir(),
task.node.uuid))
def _dhcp_option_file_or_url(task, urlboot=False, ip_version=None): def _dhcp_option_file_or_url(task, urlboot=False, ip_version=None):
@ -567,9 +553,11 @@ def dhcp_options_for_instance(task, ipxe_enabled=False, url_boot=False,
# from it's own path, but Petitboot needs it to be specified by this # from it's own path, but Petitboot needs it to be specified by this
# option since it doesn't use pxelinux.0 loader. # option since it doesn't use pxelinux.0 loader.
if not url_boot: if not url_boot:
# Enforce trailing slash
prefix = os.path.join(CONF.pxe.tftp_root, '')
dhcp_opts.append( dhcp_opts.append(
{'opt_name': DHCP_TFTP_PATH_PREFIX, {'opt_name': DHCP_TFTP_PATH_PREFIX,
'opt_value': get_tftp_path_prefix()}) 'opt_value': prefix})
if not url_boot: if not url_boot:
dhcp_opts.append({'opt_name': DHCP_TFTP_SERVER_NAME, dhcp_opts.append({'opt_name': DHCP_TFTP_SERVER_NAME,
@ -600,23 +588,6 @@ def dhcp_options_for_instance(task, ipxe_enabled=False, url_boot=False,
return dhcp_opts return dhcp_opts
def get_tftp_path_prefix():
"""Adds trailing slash (if needed) necessary for path-prefix
:return: CONF.pxe.tftp_root ensured to have a trailing slash
"""
return os.path.join(CONF.pxe.tftp_root, '')
def get_path_relative_to_tftp_root(file_path):
"""Return file relative path to CONF.pxe.tftp_root
:param file_path: full file path to be made relative path.
:returns: The path relative to CONF.pxe.tftp_root
"""
return os.path.relpath(file_path, get_tftp_path_prefix())
def is_ipxe_enabled(task): def is_ipxe_enabled(task):
"""Return true if ipxe is set. """Return true if ipxe is set.
@ -681,10 +652,7 @@ def get_instance_image_info(task, ipxe_enabled=False):
if (node.driver_internal_info.get('is_whole_disk_image') if (node.driver_internal_info.get('is_whole_disk_image')
or deploy_utils.get_boot_option(node) == 'local'): or deploy_utils.get_boot_option(node) == 'local'):
return image_info return image_info
if ipxe_enabled: root_dir = _get_root_dir(ipxe_enabled)
root_dir = get_ipxe_root_dir()
else:
root_dir = get_root_dir()
i_info = node.instance_info i_info = node.instance_info
if i_info.get('boot_iso'): if i_info.get('boot_iso'):
image_info['boot_iso'] = ( image_info['boot_iso'] = (
@ -782,8 +750,8 @@ def build_deploy_pxe_options(task, pxe_info, mode='deploy',
pxe_opts[option] = '/'.join([CONF.deploy.http_url, node.uuid, pxe_opts[option] = '/'.join([CONF.deploy.http_url, node.uuid,
label]) label])
else: else:
pxe_opts[option] = get_path_relative_to_tftp_root( pxe_opts[option] = os.path.relpath(pxe_info[label][1],
pxe_info[label][1]) CONF.pxe.tftp_root)
if ipxe_enabled: if ipxe_enabled:
pxe_opts['initrd_filename'] = ramdisk_label pxe_opts['initrd_filename'] = ramdisk_label
return pxe_opts return pxe_opts
@ -811,8 +779,8 @@ def build_instance_pxe_options(task, pxe_info, ipxe_enabled=False):
# image_source to determine if it's a whole disk image or not. # image_source to determine if it's a whole disk image or not.
# For example, when transitioning to 'available' state # For example, when transitioning to 'available' state
# for first time from 'manage' state. # for first time from 'manage' state.
pxe_opts[option] = get_path_relative_to_tftp_root( pxe_opts[option] = os.path.relpath(pxe_info[label][1],
pxe_info[label][1]) CONF.pxe.tftp_root)
pxe_opts.setdefault('aki_path', 'no_kernel') pxe_opts.setdefault('aki_path', 'no_kernel')
pxe_opts.setdefault('ari_path', 'no_ramdisk') pxe_opts.setdefault('ari_path', 'no_ramdisk')
@ -942,14 +910,6 @@ def build_service_pxe_config(task, instance_image_info,
ipxe_enabled=ipxe_enabled, anaconda_boot=anaconda_boot) ipxe_enabled=ipxe_enabled, anaconda_boot=anaconda_boot)
def _build_heartbeat_url(node_uuid):
api_version = 'v1'
heartbeat_api = '%s/heartbeat/{node_uuid}' % api_version
path = heartbeat_api.format(node_uuid=node_uuid)
return "/".join([deploy_utils.get_ironic_api_url(), path])
def build_kickstart_config_options(task): def build_kickstart_config_options(task):
"""Build the kickstart template options for a node """Build the kickstart template options for a node
@ -969,7 +929,12 @@ def build_kickstart_config_options(task):
node.save() node.save()
params['liveimg_url'] = node.instance_info['image_url'] params['liveimg_url'] = node.instance_info['image_url']
params['agent_token'] = node.driver_internal_info['agent_secret_token'] params['agent_token'] = node.driver_internal_info['agent_secret_token']
params['heartbeat_url'] = _build_heartbeat_url(node.uuid)
heartbeat_url = '%s/v1/heartbeat/%s' % (
deploy_utils.get_ironic_api_url().rstrip('/'),
node.uuid
)
params['heartbeat_url'] = heartbeat_url
return {'ks_options': params} return {'ks_options': params}
@ -1221,9 +1186,9 @@ def cache_ramdisk_kernel(task, pxe_info, ipxe_enabled=False):
node = task.node node = task.node
t_pxe_info = copy.copy(pxe_info) t_pxe_info = copy.copy(pxe_info)
if ipxe_enabled: if ipxe_enabled:
path = os.path.join(get_ipxe_root_dir(), node.uuid) path = os.path.join(CONF.deploy.http_root, node.uuid)
else: else:
path = os.path.join(get_root_dir(), node.uuid) path = os.path.join(CONF.pxe.tftp_root, node.uuid)
fileutils.ensure_tree(path) fileutils.ensure_tree(path)
# anconda deploy will have 'stage2' as one of the labels in pxe_info dict # anconda deploy will have 'stage2' as one of the labels in pxe_info dict
if 'stage2' in pxe_info.keys(): if 'stage2' in pxe_info.keys():
@ -1232,7 +1197,7 @@ def cache_ramdisk_kernel(task, pxe_info, ipxe_enabled=False):
fileutils.ensure_tree( fileutils.ensure_tree(
get_file_path_from_label( get_file_path_from_label(
node.uuid, node.uuid,
get_ipxe_root_dir(), CONF.deploy.http_root,
'stage2' 'stage2'
) )
) )

29
ironic/tests/unit/common/test_pxe_utils.py
View File

@ -764,11 +764,6 @@ class TestPXEUtils(db_base.DbTestCase):
self.assertEqual('/tftpboot/10.10.0.1.conf', self.assertEqual('/tftpboot/10.10.0.1.conf',
pxe_utils._get_pxe_ip_address_path(ipaddress)) pxe_utils._get_pxe_ip_address_path(ipaddress))
def test_get_root_dir(self):
expected_dir = '/tftproot'
self.config(tftp_root=expected_dir, group='pxe')
self.assertEqual(expected_dir, pxe_utils.get_root_dir())
def test_get_pxe_config_file_path(self): def test_get_pxe_config_file_path(self):
self.assertEqual(os.path.join(CONF.pxe.tftp_root, self.assertEqual(os.path.join(CONF.pxe.tftp_root,
self.node.uuid, self.node.uuid,
@ -794,7 +789,7 @@ class TestPXEUtils(db_base.DbTestCase):
group='pxe') group='pxe')
else: else:
self.config(pxe_bootfile_name='fake-bootfile', group='pxe') self.config(pxe_bootfile_name='fake-bootfile', group='pxe')
self.config(tftp_root='/tftp-path/', group='pxe') self.config(tftp_root='/tftp-path', group='pxe')
if ipxe: if ipxe:
bootfile = 'fake-bootfile-ipxe' bootfile = 'fake-bootfile-ipxe'
else: else:
@ -980,28 +975,6 @@ class TestPXEUtils(db_base.DbTestCase):
rmtree_mock.assert_called_once_with( rmtree_mock.assert_called_once_with(
os.path.join(CONF.pxe.tftp_root, self.node.uuid)) os.path.join(CONF.pxe.tftp_root, self.node.uuid))
def test_get_tftp_path_prefix_with_trailing_slash(self):
self.config(tftp_root='/tftpboot-path/', group='pxe')
path_prefix = pxe_utils.get_tftp_path_prefix()
self.assertEqual(path_prefix, '/tftpboot-path/')
def test_get_tftp_path_prefix_without_trailing_slash(self):
self.config(tftp_root='/tftpboot-path', group='pxe')
path_prefix = pxe_utils.get_tftp_path_prefix()
self.assertEqual(path_prefix, '/tftpboot-path/')
def test_get_path_relative_to_tftp_root_with_trailing_slash(self):
self.config(tftp_root='/tftpboot-path/', group='pxe')
test_file_path = '/tftpboot-path/pxelinux.cfg/test'
relpath = pxe_utils.get_path_relative_to_tftp_root(test_file_path)
self.assertEqual(relpath, 'pxelinux.cfg/test')
def test_get_path_relative_to_tftp_root_without_trailing_slash(self):
self.config(tftp_root='/tftpboot-path', group='pxe')
test_file_path = '/tftpboot-path/pxelinux.cfg/test'
relpath = pxe_utils.get_path_relative_to_tftp_root(test_file_path)
self.assertEqual(relpath, 'pxelinux.cfg/test')
@mock.patch('ironic.common.utils.rmtree_without_raise', autospec=True) @mock.patch('ironic.common.utils.rmtree_without_raise', autospec=True)
@mock.patch('ironic_lib.utils.unlink_without_raise', autospec=True) @mock.patch('ironic_lib.utils.unlink_without_raise', autospec=True)
@mock.patch('ironic.common.dhcp_factory.DHCPFactory.provider', @mock.patch('ironic.common.dhcp_factory.DHCPFactory.provider',