RBD: ensure temporary config gets deleted

This ensures that if a failure occurs with any of the linuxrbd calls,
the temp config file is still removed.

This also gives it a prefix so it is identifiable.

Change-Id: Ief3d907092e77c48a531a6ae78b10c58cc6db56c
changes/08/396708/3
Eric Harney 6 years ago
parent eca5a7f2e0
commit a88f13a1e4
  1. 19
      os_brick/initiator/connectors/rbd.py
  2. 2
      os_brick/tests/initiator/connectors/test_rbd.py

@ -18,6 +18,7 @@ import tempfile
from oslo_concurrency import processutils as putils
from oslo_log import log as logging
from oslo_utils import fileutils
from oslo_utils import netutils
from os_brick.i18n import _, _LE
@ -78,7 +79,7 @@ class RBDConnector(base.BaseLinuxConnector):
keyring = ("keyring = /etc/ceph/%s.client.%s.keyring" %
(cluster_name, user))
try:
fd, ceph_conf_path = tempfile.mkstemp()
fd, ceph_conf_path = tempfile.mkstemp(prefix="brickrbd_")
with os.fdopen(fd, 'w') as conf_file:
conf_file.writelines([mon_hosts, "\n",
client_section, "\n", keyring])
@ -100,14 +101,14 @@ class RBDConnector(base.BaseLinuxConnector):
conf = self._create_ceph_conf(monitor_ips, monitor_ports,
str(cluster_name), user)
rbd_client = linuxrbd.RBDClient(user, pool, conffile=conf,
rbd_cluster_name=str(cluster_name))
rbd_volume = linuxrbd.RBDVolume(rbd_client, volume)
rbd_handle = linuxrbd.RBDVolumeIOWrapper(
linuxrbd.RBDImageMetadata(rbd_volume, pool, user, conf))
if os.path.exists(conf):
os.remove(conf)
try:
rbd_client = linuxrbd.RBDClient(user, pool, conffile=conf,
rbd_cluster_name=str(cluster_name))
rbd_volume = linuxrbd.RBDVolume(rbd_client, volume)
rbd_handle = linuxrbd.RBDVolumeIOWrapper(
linuxrbd.RBDImageMetadata(rbd_volume, pool, user, conf))
finally:
fileutils.delete_if_exists(conf)
return rbd_handle

@ -124,7 +124,7 @@ class RBDConnectorTestCase(test_connector.ConnectorTestCase):
conf_path = rbd_connector._create_ceph_conf(
self.hosts, self.ports, self.clustername, self.user)
self.assertEqual(conf_path, tmpfile)
mock_mkstemp.assert_called_once_with()
mock_mkstemp.assert_called_once_with(prefix='brickrbd_')
@mock.patch.object(priv_rootwrap, 'execute', return_value=None)
def test_connect_local_volume(self, mock_execute):

Loading…
Cancel
Save