openstack/ironic
Code Issues Proposed changes

Merge "Reduce the number of small functions in pxe_utils"

Browse Source
This commit is contained in:
Zuul 2021-11-17 11:36:36 +00:00 committed by Gerrit Code Review
commit f90b5a6c11
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}
def get_root_dir():
"""Returns the directory where the config files and images will live."""
return CONF.pxe.tftp_root
def get_ipxe_root_dir():
return CONF.deploy.http_root
def _get_root_dir(ipxe_enabled):
if ipxe_enabled:
return CONF.deploy.http_root
else:
return CONF.pxe.tftp_root
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
is in use by the caller.
"""
if ipxe_enabled:
root_dir = get_ipxe_root_dir()
else:
root_dir = get_root_dir()
root_dir = _get_root_dir(ipxe_enabled)
node_dir = os.path.join(root_dir, task.node.uuid)
pxe_dir = os.path.join(root_dir, PXE_CFG_DIR_NAME)
# 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):
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" %
("grub.cfg", "01", mac.replace(':', "-").lower()))
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:
hw_type = '20-'
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)
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)
@ -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.
"""
if ipxe_enabled:
root_dir = get_ipxe_root_dir()
else:
root_dir = get_root_dir()
root_dir = _get_root_dir(ipxe_enabled)
image_info = {}
labels = KERNEL_RAMDISK_LABELS[mode]
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.
"""
if ipxe_enabled:
return os.path.join(get_ipxe_root_dir(), node_uuid, 'config')
else:
return os.path.join(get_root_dir(), node_uuid, 'config')
return os.path.join(_get_root_dir(ipxe_enabled), node_uuid, 'config')
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
"""
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':
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':
return os.path.join(get_ipxe_root_dir(), node_uuid, 'LiveOS',
return os.path.join(CONF.deploy.http_root, node_uuid, 'LiveOS',
'squashfs.img')
else:
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,
ipxe_enabled=ipxe_enabled):
ironic_utils.unlink_without_raise(path)
if ipxe_enabled:
utils.rmtree_without_raise(os.path.join(get_ipxe_root_dir(),
task.node.uuid))
else:
utils.rmtree_without_raise(os.path.join(get_root_dir(),
task.node.uuid))
utils.rmtree_without_raise(os.path.join(_get_root_dir(ipxe_enabled),
task.node.uuid))
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
# option since it doesn't use pxelinux.0 loader.
if not url_boot:
# Enforce trailing slash
prefix = os.path.join(CONF.pxe.tftp_root, '')
dhcp_opts.append(
{'opt_name': DHCP_TFTP_PATH_PREFIX,
'opt_value': get_tftp_path_prefix()})
'opt_value': prefix})
if not url_boot:
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
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):
"""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')
or deploy_utils.get_boot_option(node) == 'local'):
return image_info
if ipxe_enabled:
root_dir = get_ipxe_root_dir()
else:
root_dir = get_root_dir()
root_dir = _get_root_dir(ipxe_enabled)
i_info = node.instance_info
if i_info.get('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,
label])
else:
pxe_opts[option] = get_path_relative_to_tftp_root(
pxe_info[label][1])
pxe_opts[option] = os.path.relpath(pxe_info[label][1],
CONF.pxe.tftp_root)
if ipxe_enabled:
pxe_opts['initrd_filename'] = ramdisk_label
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.
# For example, when transitioning to 'available' state
# for first time from 'manage' state.
pxe_opts[option] = get_path_relative_to_tftp_root(
pxe_info[label][1])
pxe_opts[option] = os.path.relpath(pxe_info[label][1],
CONF.pxe.tftp_root)
pxe_opts.setdefault('aki_path', 'no_kernel')
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)
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):
"""Build the kickstart template options for a node
@ -969,7 +929,12 @@ def build_kickstart_config_options(task):
node.save()
params['liveimg_url'] = node.instance_info['image_url']
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}
@ -1221,9 +1186,9 @@ def cache_ramdisk_kernel(task, pxe_info, ipxe_enabled=False):
node = task.node
t_pxe_info = copy.copy(pxe_info)
if ipxe_enabled:
path = os.path.join(get_ipxe_root_dir(), node.uuid)
path = os.path.join(CONF.deploy.http_root, node.uuid)
else:
path = os.path.join(get_root_dir(), node.uuid)
path = os.path.join(CONF.pxe.tftp_root, node.uuid)
fileutils.ensure_tree(path)
# anconda deploy will have 'stage2' as one of the labels in pxe_info dict
if 'stage2' in pxe_info.keys():
@ -1232,7 +1197,7 @@ def cache_ramdisk_kernel(task, pxe_info, ipxe_enabled=False):
fileutils.ensure_tree(
get_file_path_from_label(
node.uuid,
get_ipxe_root_dir(),
CONF.deploy.http_root,
'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',
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):
self.assertEqual(os.path.join(CONF.pxe.tftp_root,
self.node.uuid,
@ -794,7 +789,7 @@ class TestPXEUtils(db_base.DbTestCase):
group='pxe')
else:
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:
bootfile = 'fake-bootfile-ipxe'
else:
@ -980,28 +975,6 @@ class TestPXEUtils(db_base.DbTestCase):
rmtree_mock.assert_called_once_with(
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_lib.utils.unlink_without_raise', autospec=True)
@mock.patch('ironic.common.dhcp_factory.DHCPFactory.provider',