Nir Soffer has uploaded a new change for review.
Change subject: misc: Safer and simpler itmap
......................................................................
misc: Safer and simpler itmap
The previous code had few issues:
- It used unlimited number of threads by default. This may lead to
creation of 100's of threads if you do not specify a value.
- It used non-daemon threads, which could lead to unwanted delay during
vdsm shutdown.
- It tried to yield results before all arguments were handled. This
could lead to unwanted delay in argument processing, if the caller
would block processing the results.
- It started one thread per value, even if maxthreads was smaller than
number of values.
- It was too complicated.
Changes:
- The caller must specify the maximum number of threads.
- Use daemon threads
- Queue all values before yielding results
- Start up to maxthreads worker threads, each processing multiple values
- Simplify the code
- Add test for error handling
Change-Id: Iba6116ac4003702c8e921cebaf494491a6f9afaf
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com>
---
M tests/miscTests.py
M vdsm/storage/misc.py
2 files changed, 42 insertions(+), 42 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/19/39119/1
diff --git a/tests/miscTests.py b/tests/miscTests.py
index 31f64fa..4b3e3c3 100644
--- a/tests/miscTests.py
+++ b/tests/miscTests.py
@@ -196,7 +196,7 @@
# outOfProcess operation + 1. it let us know that oop and itmap operate
# properly with their limitations
data = frozenset(range(oop.HELPERS_PER_DOMAIN + 1))
- ret = frozenset(misc.itmap(dummy, data, misc.UNLIMITED_THREADS))
+ ret = frozenset(misc.itmap(dummy, data, len(data)))
self.assertEquals(ret, data)
def testMoreThreadsThanArgs(self):
@@ -207,6 +207,13 @@
data = 1
self.assertRaises(ValueError, misc.itmap(int, data, 0).next)
+ def testErrors(self):
+ err = Exception()
+ def dummy(arg):
+ raise err
+ data = [1, 2, 3]
+ self.assertEqual(list(misc.itmap(dummy, data, 4)), [err] * len(data))
+
class RotateFiles(TestCaseBase):
diff --git a/vdsm/storage/misc.py b/vdsm/storage/misc.py
index eb484c7..463fd04 100644
--- a/vdsm/storage/misc.py
+++ b/vdsm/storage/misc.py
@@ -58,7 +58,6 @@
STR_UUID_SIZE = 36
UUID_HYPHENS = [8, 13, 18, 23]
MEGA = 1 << 20
-UNLIMITED_THREADS = -1
log = logging.getLogger('Storage.Misc')
@@ -882,53 +881,47 @@
raise exception
-def itmap(func, iterable, maxthreads=UNLIMITED_THREADS):
+def itmap(func, iterable, maxthreads):
"""
- Make an iterator that computes the function using
- arguments from the iterable. It works similar to tmap
- by running each operation in a different thread, this
- causes the results not to return in any particular
- order so it's good if you don't care about the order
- of the results.
- maxthreads stands for maximum threads that we can initiate simultaneosly.
- If we reached to max threads the function waits for thread to
- finish before initiate the next one.
+ Return an iterator calling func with arguments from iterable in multiple threads.
+
+ Unlike tmap, the results are not returned in the original order of the
+ arguments, and number of threads is limited to maxthreads.
"""
- if maxthreads < 1 and maxthreads != UNLIMITED_THREADS:
- raise ValueError("Wrong input to function itmap: %s", maxthreads)
+ if maxthreads < 1:
+ raise ValueError("Invalid maxthreads value: %s" % maxthreads)
- respQueue = Queue.Queue()
+ DONE = object()
+ values = Queue.Queue()
+ results = Queue.Queue()
- def wrapper(value):
- try:
- respQueue.put(func(value))
- except Exception as e:
- respQueue.put(e)
+ def worker():
+ while True:
+ value = values.get()
+ if value is DONE:
+ return
+ try:
+ results.put(func(value))
+ except Exception as e:
+ results.put(e)
- threadsCount = 0
- for arg in iterable:
- if maxthreads != UNLIMITED_THREADS:
- if maxthreads == 0:
- # This not supposed to happened. If it does, it's a bug.
- # maxthreads should get to 0 only after threadsCount is
- # greater than 1
- if threadsCount < 1:
- raise RuntimeError("No thread initiated")
- else:
- yield respQueue.get()
- # if yield returns one thread stopped, so we can run
- # another thread in queue
- maxthreads += 1
- threadsCount -= 1
+ count = 0
+ threads = 0
- t = threading.Thread(target=wrapper, args=(arg,))
- t.start()
- threadsCount += 1
- maxthreads -= 1
+ for value in iterable:
+ values.put(value)
+ count += 1
+ if threads < maxthreads:
+ t = threading.Thread(target=worker)
+ t.daemon = True
+ t.start()
+ threads += 1
- # waiting for rest threads to end
- for i in xrange(threadsCount):
- yield respQueue.get()
+ for _ in range(threads):
+ values.put(DONE)
+
+ for _ in xrange(count):
+ yield results.get()
def isAscii(s):
--
To view, visit https://gerrit.ovirt.org/39119
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Iba6116ac4003702c8e921cebaf494491a6f9afaf
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com>
Nir Soffer has uploaded a new change for review.
Change subject: udevadm: More precise error handling
......................................................................
udevadm: More precise error handling
udevadm provides a --timeout option, but there is no robust way to
detect a timeout in EL6, EL7, and Fedora 20. In Fedora 21 and upstream,
udevadm ignores the timeout option. This patch improves error handling
by using our own timeout.
udevadm.settle() raises now udevadm.Failure or udevadm.Timeout, and the
caller is responsible to handle the error.
In both multipath.rescan() and IscsiConnection.connect(), we warn about
timeout but do not handle other errors, so real errors in udevadm will
fail loudly.
Change-Id: Ia0a7380b1b181ec93399ea741122cfa2e98086fb
Relates-To: https://bugzilla.redhat.com/1209474
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com>
---
A tests/udevadmTests.py
M vdsm/storage/multipath.py
M vdsm/storage/storageServer.py
M vdsm/storage/udevadm.py
4 files changed, 106 insertions(+), 21 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/40/39740/1
diff --git a/tests/udevadmTests.py b/tests/udevadmTests.py
new file mode 100644
index 0000000..90841b2
--- /dev/null
+++ b/tests/udevadmTests.py
@@ -0,0 +1,52 @@
+#
+# Copyright 2015 Red Hat, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+#
+# Refer to the README and COPYING files for full details of the license
+#
+
+
+from monkeypatch import MonkeyPatch
+from testlib import VdsmTestCase
+
+from vdsm import utils
+from storage import udevadm
+
+TRUE = utils.CommandPath("true", "/bin/true", "/usr/bin/true")
+FALSE = utils.CommandPath("false", "/bin/false", "/usr/bin/false")
+READ = utils.CommandPath("read", "/bin/read", "/usr/bin/read")
+
+
+class UdevadmSettleTests(VdsmTestCase):
+
+ @MonkeyPatch(udevadm, "_UDEVADM", TRUE)
+ def test_success(self):
+ udevadm.settle(5)
+
+ @MonkeyPatch(udevadm, "_UDEVADM", FALSE)
+ def test_error(self):
+ try:
+ udevadm.settle(5)
+ except udevadm.Failure as e:
+ self.assertEqual(e.rc, 1)
+ self.assertEqual(e.out, "")
+ self.assertEqual(e.err, "")
+ else:
+ self.fail("Failure not raised")
+
+ @MonkeyPatch(udevadm, "_UDEVADM", READ)
+ def test_timeout(self):
+ self.assertRaises(udevadm.Timeout, udevadm.settle, 1)
diff --git a/vdsm/storage/multipath.py b/vdsm/storage/multipath.py
index a1c42b3..925c411 100644
--- a/vdsm/storage/multipath.py
+++ b/vdsm/storage/multipath.py
@@ -73,7 +73,10 @@
# events are processed, ensuring detection of new devices and creation or
# update of multipath devices.
timeout = config.getint('irs', 'scsi_settle_timeout')
- udevadm.settle(timeout)
+ try:
+ udevadm.settle(timeout)
+ except udevadm.Timeout as e:
+ log.warning("Timeout waiting for udev events: %s", e)
def deduceType(a, b):
diff --git a/vdsm/storage/storageServer.py b/vdsm/storage/storageServer.py
index 22a90d1..c19fb8d 100644
--- a/vdsm/storage/storageServer.py
+++ b/vdsm/storage/storageServer.py
@@ -382,7 +382,10 @@
def connect(self):
iscsi.addIscsiNode(self._iface, self._target, self._cred)
timeout = config.getint("irs", "scsi_settle_timeout")
- udevadm.settle(timeout)
+ try:
+ udevadm.settle(timeout)
+ except udevadm.Timeout as e:
+ self.log.warning("Timeout waiting for udev events: %s", e)
def _match(self, session):
target = session.target
diff --git a/vdsm/storage/udevadm.py b/vdsm/storage/udevadm.py
index 4b4b54a..a2afd04 100644
--- a/vdsm/storage/udevadm.py
+++ b/vdsm/storage/udevadm.py
@@ -18,22 +18,39 @@
# Refer to the README and COPYING files for full details of the license
#
-import logging
+import errno
+import signal
+
from vdsm import utils
+from vdsm.infra import zombiereaper
_UDEVADM = utils.CommandPath("udevadm", "/sbin/udevadm", "/usr/sbin/udevadm")
class Error(Exception):
+ message = None
- def __init__(self, rc, out, err):
+ def __str__(self):
+ return self.message.format(self=self)
+
+
+class Failure(Error):
+ message = ("udevadm failed cmd={self.cmd} rc={self.rc} out={self.out!r} "
+ "err={self.err!r}")
+
+ def __init__(self, cmd, rc, out, err):
+ self.cmd = cmd
self.rc = rc
self.out = out
self.err = err
- def __str__(self):
- return "Process failed with rc=%d out=%r err=%r" % (
- self.rc, self.out, self.err)
+
+class Timeout(Error):
+ message = ("udevadm timed out cmd={self.cmd} timeout={self.timeout}")
+
+ def __init__(self, cmd, timeout):
+ self.cmd = cmd
+ self.timeout = timeout
def settle(timeout, exit_if_exists=None):
@@ -44,25 +61,35 @@
Arguments:
timeout Maximum number of seconds to wait for the event queue to
- become empty. A value of 0 will check if the queue is empty
- and always return immediately.
+ become empty.
exit_if_exists Stop waiting if file exists.
+
+ Raises Failure if udevadm failed, or Timeout if udevadm did not terminate
+ within the requested timeout.
"""
- args = ["settle", "--timeout=%s" % timeout]
+ cmd = [_UDEVADM.cmd, "settle"]
if exit_if_exists:
- args.append("--exit-if-exists=%s" % exit_if_exists)
+ cmd.append("--exit-if-exists=%s" % exit_if_exists)
- try:
- _run_command(args)
- except Error as e:
- logging.error("%s", e)
+ _run_command(cmd, timeout)
-def _run_command(args):
- cmd = [_UDEVADM.cmd]
- cmd.extend(args)
- rc, out, err = utils.execCmd(cmd, raw=True)
- if rc != 0:
- raise Error(rc, out, err)
+def _run_command(cmd, timeout=None):
+ proc = utils.execCmd(cmd, sync=False, deathSignal=signal.SIGKILL)
+
+ if not proc.wait(timeout):
+ try:
+ proc.kill()
+ except OSError as e:
+ if e.errno != errno.ESRCH:
+ raise
+ finally:
+ zombiereaper.autoReapPID(proc.pid)
+ raise Timeout(cmd, timeout)
+
+ if proc.returncode != 0:
+ out = "".join(proc.stdout)
+ err = "".join(proc.stderr)
+ raise Failure(cmd, proc.returncode, out, err)
--
To view, visit https://gerrit.ovirt.org/39740
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ia0a7380b1b181ec93399ea741122cfa2e98086fb
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com>
Roman Mohr has uploaded a new change for review.
Change subject: monitor: Add udev monitor
......................................................................
monitor: Add udev monitor
Allow easy monitoring of device hotplug and device state changes which
are covered by udev.
Change-Id: I4b91753424d83896fa538eb6b57f8653b6332fbb
Signed-off-by: Roman Mohr <rmohr(a)redhat.com>
---
A lib/vdsm/udev/monitor.py
M tests/Makefile.am
A tests/udevMonitorTests.py
M vdsm.spec.in
4 files changed, 272 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/29/47729/1
diff --git a/lib/vdsm/udev/monitor.py b/lib/vdsm/udev/monitor.py
new file mode 100644
index 0000000..8698bf3
--- /dev/null
+++ b/lib/vdsm/udev/monitor.py
@@ -0,0 +1,133 @@
+# Copyright 2015 Red Hat, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+#
+# Refer to the README and COPYING files for full details of the license
+#
+from functools import partial
+import logging
+import threading
+from time import sleep
+
+from pyudev import Context, Monitor, MonitorObserver
+
+
+class UdevMonitor(object):
+
+ """
+ ``udev`` event monitor. Usage:
+
+ The monitor is a thread-save thin wrapper arount pyudev.MonitorObserver.
+ This allows multiple callbacks for the same netlink socket. To avoid
+ listening for udev events the application is not interested in, the
+ monitoring thread only starts listening on the socket when the monitor is
+ started and at least one subscriber is added.
+
+ The simplest way to use the monitor is to subscribe a callback to a
+ specific subsystem event and let the callback do the work:
+
+ dev listen_for_disabled_cpu_events(device):
+ if device.action == 'offline':
+ print('CPU {0.name} is now offline'.format(device))
+
+ monitor = UdevMonitor()
+ monitor.start()
+ monitor.subscribe(listen_for_disabled_cpu_events, subsystem='cpu')
+
+ Another approach would be to just enqueue the udev event in a queue and do
+ the actual work in another thread or a thread pool. This is one approach to
+ do this:
+
+ queue = Queue.Queue()
+ def new_device_listener(device):
+ if device.action == 'add':
+ queue.put(device)
+
+ monitor = UdevMonitor()
+ monitor.subscribe(new_device_listener, subsystem='usb',
+ device_type='usb_device')
+ monitor.start()
+ """
+
+ def __init__(self):
+ self._subsystems = {}
+ self._context = Context()
+ self._monitor = Monitor.from_netlink(self._context)
+ self._observer = MonitorObserver(self._monitor, callback=partial(
+ UdevMonitor._event_loop, self), name='udev-monitor')
+ self._filter_lock = threading.Lock()
+ self._is_started = False
+ self._can_start = False
+
+ def _event_loop(self, device):
+ subsystem = self._subsystems[device.subsystem]
+ for callback in subsystem.get(device.device_type, []):
+ _execute_callback(callback, device)
+ if device.device_type:
+ for callback in subsystem.get(None, []):
+ _execute_callback(callback, device)
+
+ def subscribe(self, callback, subsystem, device_type=None):
+ """
+ Raise :exc:`~exceptions.ValueError` if the callback is None
+
+ :param callback: function to invoke
+ :param subsystem: byte or unicode string representing the subsystem to
+ listen on (e.g. ``'cpu'``, ``'usb'``)
+ :param device_type: byte or unicode string representing the device type
+ to listen for changes (e.g. ``'usb_device'``,
+ ``'block'``)
+ :return: None
+ """
+ if callback is None:
+ raise ValueError('callback missing')
+ with self._filter_lock:
+ self._monitor.filter_by(subsystem, device_type)
+ device_types = self._subsystems.get(subsystem, {})
+ callbacks = device_types.get(device_type, [])
+ callbacks.append(callback)
+ device_types[device_type] = callbacks
+ self._subsystems[subsystem] = device_types
+ self._start_if_necessary()
+
+ def start(self):
+ self._can_start = True
+ with self._filter_lock:
+ if self._subsystems:
+ self._start_if_necessary()
+
+ def _start_if_necessary(self):
+ if self._can_start and not self._is_started:
+ self._observer.start()
+ self._is_started = True
+
+ def stop(self):
+ """
+ Stops the monitoring thread. It is guaranteed that callbacks are no
+ longer invoked after calling this method. The method can be called
+ multiple times.
+
+ Note that this only stops the monitoring thread and not the monitor
+ itself. The monitor stops listening when it is dereferenced.
+ """
+ self._observer.stop()
+
+
+def _execute_callback(callback, device):
+ try:
+ callback(device)
+ except Exception as callbackException:
+ logging.error(
+ 'Callback execution threw an exception: %s', callbackException)
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 1168862..7483d0c 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -110,6 +110,7 @@
testlibTests.py \
toolTests.py \
transportWrapperTests.py \
+ udevMonitorTests.py \
utilsTests.py \
vdscliTests.py \
vdsClientTests.py \
diff --git a/tests/udevMonitorTests.py b/tests/udevMonitorTests.py
new file mode 100644
index 0000000..2526459
--- /dev/null
+++ b/tests/udevMonitorTests.py
@@ -0,0 +1,137 @@
+# Copyright 2015 Red Hat, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+#
+# Refer to the README and COPYING files for full details of the license
+#
+
+from functools import partial
+
+import Queue
+
+from nettestlib import Bridge
+from testValidation import ValidateRunningAsRoot
+from testlib import VdsmTestCase
+from vdsm.udev.monitor import UdevMonitor
+
+
+class UdevMonitorTest(VdsmTestCase):
+
+ def setUp(self):
+ self._queue = Queue.Queue()
+ self._monitor = UdevMonitor()
+ self._monitor.start()
+
+ def tearDown(self):
+ self._monitor.stop()
+
+ @ValidateRunningAsRoot
+ def testAddDevice(self):
+ self._monitor.subscribe(
+ partial(UdevMonitorTest._device_listener, self),
+ subsystem='net',
+ device_type='bridge')
+ bridge = Bridge()
+ try:
+ bridge.addDevice()
+ device = None
+ try:
+ device = self._queue.get(timeout=1)
+ except Queue.Empty:
+ pass
+ self.assertIsNotNone(device,
+ msg='Should have detected an event')
+ self.assertEqual(device.action, 'add')
+ finally:
+ bridge.delDevice()
+
+ @ValidateRunningAsRoot
+ def testAddDeviceForSubsystem(self):
+ self._monitor.subscribe(
+ partial(UdevMonitorTest._device_listener, self),
+ subsystem='net',
+ device_type=None)
+ bridge = Bridge()
+ try:
+ bridge.addDevice()
+ device = None
+ try:
+ device = self._queue.get(timeout=1)
+ except Queue.Empty:
+ pass
+ self.assertIsNotNone(device,
+ msg='Should have detected an event')
+ self.assertEqual(device.action, 'add')
+ finally:
+ bridge.delDevice()
+
+ @ValidateRunningAsRoot
+ def testRemoveDevice(self):
+
+ bridge = Bridge()
+ try:
+ bridge.addDevice()
+ self._monitor.subscribe(
+ partial(UdevMonitorTest._device_listener, self),
+ subsystem='net',
+ device_type='bridge')
+ bridge.delDevice()
+ device = None
+ try:
+ device = self._queue.get(timeout=1)
+ except Queue.Empty:
+ pass
+ self.assertIsNotNone(device,
+ msg='Should have detected an event')
+ self.assertEqual(device.action, 'remove')
+ except:
+ bridge.delDevice()
+
+ @ValidateRunningAsRoot
+ def testEventSeries(self):
+
+ bridge1 = Bridge()
+ bridge2 = Bridge()
+ try:
+ self._monitor.subscribe(
+ partial(UdevMonitorTest._device_listener, self),
+ subsystem='net',
+ device_type='bridge')
+ bridge1.addDevice()
+ bridge1.delDevice()
+ bridge2.addDevice()
+ bridge2.delDevice()
+ events = []
+ try:
+ events.append(self._queue.get(timeout=1))
+ events.append(self._queue.get(timeout=1))
+ events.append(self._queue.get(timeout=1))
+ events.append(self._queue.get(timeout=1))
+ except Queue.Empty:
+ pass
+ self.assertEqual(len(events), 4)
+ except:
+ _delBridge(bridge1)
+ _delBridge(bridge2)
+
+ def _device_listener(self, device):
+ self._queue.put(device)
+
+
+def _delBridge(bridge):
+ try:
+ bridge.delDevice()
+ except:
+ pass
diff --git a/vdsm.spec.in b/vdsm.spec.in
index 8f758ca..ae5e483 100644
--- a/vdsm.spec.in
+++ b/vdsm.spec.in
@@ -126,6 +126,7 @@
Requires: python-pthreading >= 0.1.3-3
Requires: python-six
Requires: python-requests
+Requires: python-udev >= 0.15
Requires: %{name}-infra = %{version}-%{release}
Requires: rpm-python
Requires: nfs-utils
--
To view, visit https://gerrit.ovirt.org/47729
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I4b91753424d83896fa538eb6b57f8653b6332fbb
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Roman Mohr <rmohr(a)redhat.com>
Adam Litke has uploaded a new change for review.
Change subject: VolumeMetadata: move prepare and teardown
......................................................................
VolumeMetadata: move prepare and teardown
Change-Id: Iba0954ace3b1da5ea9afc41aeb6ea69a729fe29c
Signed-off-by: Adam Litke <alitke(a)redhat.com>
---
M vdsm/storage/blockVolume.py
M vdsm/storage/fileVolume.py
M vdsm/storage/volume.py
3 files changed, 117 insertions(+), 109 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/50/44050/1
diff --git a/vdsm/storage/blockVolume.py b/vdsm/storage/blockVolume.py
index 0168d64..597e7ee 100644
--- a/vdsm/storage/blockVolume.py
+++ b/vdsm/storage/blockVolume.py
@@ -74,6 +74,8 @@
volume.VolumeMetadata.__init__(self, repoPath, sdUUID, imgUUID,
volUUID)
self.metaoff = None
+ self.lvmActivationNamespace = sd.getNamespace(self.sdUUID,
+ LVM_ACTIVATION_NAMESPACE)
def getMetadataId(self):
"""
@@ -332,6 +334,49 @@
lvs = lvm.lvsByTag(sdUUID, "%s%s" % (TAG_PREFIX_IMAGE, imgUUID))
return [lv.name for lv in lvs]
+ @logskip("ResourceManager")
+ def llPrepare(self, rw=False, setrw=False):
+ """
+ Perform low level volume use preparation
+
+ For the Block Volumes the actual LV activation is wrapped
+ into lvmActivation resource. It is being initialized by the
+ storage domain sitting on top of the encapsulating VG.
+ We just use it here.
+ """
+ if setrw:
+ self.setrw(rw=rw)
+ access = rm.LockType.exclusive if rw else rm.LockType.shared
+ activation = rmanager.acquireResource(self.lvmActivationNamespace,
+ self.volUUID, access)
+ activation.autoRelease = False
+
+ @classmethod
+ def teardown(cls, sdUUID, volUUID, justme=False):
+ """
+ Deactivate volume and release resources.
+ Volume deactivation occurs as part of resource releasing.
+ If justme is false, the entire COW chain should be torn down.
+ """
+ cls.log.info("Tearing down volume %s/%s justme %s"
+ % (sdUUID, volUUID, justme))
+ lvmActivationNamespace = sd.getNamespace(sdUUID,
+ LVM_ACTIVATION_NAMESPACE)
+ rmanager.releaseResource(lvmActivationNamespace, volUUID)
+ if not justme:
+ try:
+ pvolUUID = _getVolumeTag(sdUUID, volUUID, TAG_PREFIX_PARENT)
+ except Exception as e:
+ # If storage not accessible or lvm error occurred
+ # we will failure to get the parent volume.
+ # We can live with it and still succeed in volume's teardown.
+ pvolUUID = volume.BLANK_UUID
+ cls.log.warn("Failure to get parent of volume %s/%s (%s)"
+ % (sdUUID, volUUID, e))
+
+ if pvolUUID != volume.BLANK_UUID:
+ cls.teardown(sdUUID=sdUUID, volUUID=pvolUUID, justme=False)
+
class BlockVolume(volume.Volume):
""" Actually represents a single volume (i.e. part of virtual disk).
@@ -341,8 +386,6 @@
def __init__(self, repoPath, sdUUID, imgUUID, volUUID):
md = self.MetadataClass(repoPath, sdUUID, imgUUID, volUUID)
volume.Volume.__init__(self, md)
- self.lvmActivationNamespace = sd.getNamespace(self.sdUUID,
- LVM_ACTIVATION_NAMESPACE)
@property
def metaoff(self):
@@ -609,49 +652,6 @@
def shareVolumeRollback(cls, taskObj, volPath):
cls.log.info("Volume rollback for volPath=%s", volPath)
utils.rmFile(volPath)
-
- @logskip("ResourceManager")
- def llPrepare(self, rw=False, setrw=False):
- """
- Perform low level volume use preparation
-
- For the Block Volumes the actual LV activation is wrapped
- into lvmActivation resource. It is being initialized by the
- storage domain sitting on top of the encapsulating VG.
- We just use it here.
- """
- if setrw:
- self.setrw(rw=rw)
- access = rm.LockType.exclusive if rw else rm.LockType.shared
- activation = rmanager.acquireResource(self.lvmActivationNamespace,
- self.volUUID, access)
- activation.autoRelease = False
-
- @classmethod
- def teardown(cls, sdUUID, volUUID, justme=False):
- """
- Deactivate volume and release resources.
- Volume deactivation occurs as part of resource releasing.
- If justme is false, the entire COW chain should be torn down.
- """
- cls.log.info("Tearing down volume %s/%s justme %s"
- % (sdUUID, volUUID, justme))
- lvmActivationNamespace = sd.getNamespace(sdUUID,
- LVM_ACTIVATION_NAMESPACE)
- rmanager.releaseResource(lvmActivationNamespace, volUUID)
- if not justme:
- try:
- pvolUUID = _getVolumeTag(sdUUID, volUUID, TAG_PREFIX_PARENT)
- except Exception as e:
- # If storage not accessible or lvm error occurred
- # we will failure to get the parent volume.
- # We can live with it and still succeed in volume's teardown.
- pvolUUID = volume.BLANK_UUID
- cls.log.warn("Failure to get parent of volume %s/%s (%s)"
- % (sdUUID, volUUID, e))
-
- if pvolUUID != volume.BLANK_UUID:
- cls.teardown(sdUUID=sdUUID, volUUID=pvolUUID, justme=False)
def getVolumeTag(self, tagPrefix):
return self.md.getVolumeTag(tagPrefix)
diff --git a/vdsm/storage/fileVolume.py b/vdsm/storage/fileVolume.py
index 2dad91f..08e3f50 100644
--- a/vdsm/storage/fileVolume.py
+++ b/vdsm/storage/fileVolume.py
@@ -337,6 +337,26 @@
volList.append(volid)
return volList
+ def llPrepare(self, rw=False, setrw=False):
+ """
+ Make volume accessible as readonly (internal) or readwrite (leaf)
+ """
+ volPath = self.getVolumePath()
+
+ # Volumes leaves created in 2.2 did not have group writeable bit
+ # set. We have to set it here if we want qemu-kvm to write to old
+ # NFS volumes.
+ self.oop.fileUtils.copyUserModeToGroup(volPath)
+
+ if setrw:
+ self.setrw(rw=rw)
+ if rw:
+ if not self.oop.os.access(volPath, os.R_OK | os.W_OK):
+ raise se.VolumeAccessError(volPath)
+ else:
+ if not self.oop.os.access(volPath, os.R_OK):
+ raise se.VolumeAccessError(volPath)
+
class FileVolume(volume.Volume):
""" Actually represents a single volume (i.e. part of virtual disk).
@@ -488,26 +508,6 @@
procPool.utils.rmFile(volPath)
procPool.utils.rmFile(cls.__metaVolumePath(volPath))
procPool.utils.rmFile(cls.MetadataClass._leaseVolumePath(volPath))
-
- def llPrepare(self, rw=False, setrw=False):
- """
- Make volume accessible as readonly (internal) or readwrite (leaf)
- """
- volPath = self.getVolumePath()
-
- # Volumes leaves created in 2.2 did not have group writeable bit
- # set. We have to set it here if we want qemu-kvm to write to old
- # NFS volumes.
- self.oop.fileUtils.copyUserModeToGroup(volPath)
-
- if setrw:
- self.setrw(rw=rw)
- if rw:
- if not self.oop.os.access(volPath, os.R_OK | os.W_OK):
- raise se.VolumeAccessError(volPath)
- else:
- if not self.oop.os.access(volPath, os.R_OK):
- raise se.VolumeAccessError(volPath)
@classmethod
def __putMetadata(cls, metaId, meta):
diff --git a/vdsm/storage/volume.py b/vdsm/storage/volume.py
index a098eb7..22f7ead 100644
--- a/vdsm/storage/volume.py
+++ b/vdsm/storage/volume.py
@@ -525,6 +525,56 @@
return apparentSize
return req_size
+ def prepare(self, rw=True, justme=False,
+ chainrw=False, setrw=False, force=False):
+ """
+ Prepare volume for use by consumer.
+ If justme is false, the entire COW chain is prepared.
+ Note: setrw arg may be used only by SPM flows.
+ """
+ self.log.info("Volume: preparing volume %s/%s",
+ self.sdUUID, self.volUUID)
+
+ if not force:
+ # Cannot prepare ILLEGAL volume
+ if not self.isLegal():
+ raise se.prepareIllegalVolumeError(self.volUUID)
+
+ if rw and self.isShared():
+ if chainrw:
+ rw = False # Shared cannot be set RW
+ else:
+ raise se.SharedVolumeNonWritable(self)
+
+ if (not chainrw and rw and self.isInternal() and setrw and
+ not self.recheckIfLeaf()):
+ raise se.InternalVolumeNonWritable(self)
+
+ self.llPrepare(rw=rw, setrw=setrw)
+ self.updateInvalidatedSize()
+
+ try:
+ if justme:
+ return True
+ pvol = self.produceParent()
+ if pvol:
+ pvol.prepare(rw=chainrw, justme=False,
+ chainrw=chainrw, setrw=setrw)
+ except Exception:
+ self.log.error("Unexpected error", exc_info=True)
+ self.teardown(self.sdUUID, self.volUUID)
+ raise
+
+ return True
+
+ @classmethod
+ def teardown(cls, sdUUID, volUUID, justme=False):
+ """
+ Teardown volume.
+ If justme is false, the entire COW chain is teared down.
+ """
+ pass
+
class Volume(object):
log = logging.getLogger('Storage.Volume')
@@ -1091,53 +1141,11 @@
def prepare(self, rw=True, justme=False,
chainrw=False, setrw=False, force=False):
- """
- Prepare volume for use by consumer.
- If justme is false, the entire COW chain is prepared.
- Note: setrw arg may be used only by SPM flows.
- """
- self.log.info("Volume: preparing volume %s/%s",
- self.sdUUID, self.volUUID)
-
- if not force:
- # Cannot prepare ILLEGAL volume
- if not self.isLegal():
- raise se.prepareIllegalVolumeError(self.volUUID)
-
- if rw and self.isShared():
- if chainrw:
- rw = False # Shared cannot be set RW
- else:
- raise se.SharedVolumeNonWritable(self)
-
- if (not chainrw and rw and self.isInternal() and setrw and
- not self.recheckIfLeaf()):
- raise se.InternalVolumeNonWritable(self)
-
- self.llPrepare(rw=rw, setrw=setrw)
- self.updateInvalidatedSize()
-
- try:
- if justme:
- return True
- pvol = self.produceParent()
- if pvol:
- pvol.prepare(rw=chainrw, justme=False,
- chainrw=chainrw, setrw=setrw)
- except Exception:
- self.log.error("Unexpected error", exc_info=True)
- self.teardown(self.sdUUID, self.volUUID)
- raise
-
- return True
+ return self.md.prepare(rw, justme, chainrw, setrw, force)
@classmethod
def teardown(cls, sdUUID, volUUID, justme=False):
- """
- Teardown volume.
- If justme is false, the entire COW chain is teared down.
- """
- pass
+ cls.MetadataClass.teardown(sdUUID, volUUID, justme)
@classmethod
def newMetadata(cls, metaId, sdUUID, imgUUID, puuid, size, format, type,
--
To view, visit https://gerrit.ovirt.org/44050
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Iba0954ace3b1da5ea9afc41aeb6ea69a729fe29c
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Adam Litke <alitke(a)redhat.com>
Adam Litke has uploaded a new change for review.
Change subject: storage: Make Image.__chainSizeCalc public
......................................................................
storage: Make Image.__chainSizeCalc public
The new SDM copyVolumeData wants to make use of the same logic being
used by the classic copy flows to extend the size of the target volume
to the appropriate size. Make Image.__chainSizeCalc public so it can be
accessed from the SDM code.
Change-Id: Id079eb5067c16f934370e42b5f4e09bbcef1512b
Signed-off-by: Adam Litke <alitke(a)redhat.com>
---
M vdsm/storage/image.py
1 file changed, 2 insertions(+), 2 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/95/38995/1
diff --git a/vdsm/storage/image.py b/vdsm/storage/image.py
index 791b48c..8d5e8c2 100644
--- a/vdsm/storage/image.py
+++ b/vdsm/storage/image.py
@@ -140,7 +140,7 @@
randomStr = misc.randomStr(RENAME_RANDOM_STRING_LEN)
return "%s%s_%s" % (sd.REMOVED_IMAGE_PREFIX, randomStr, uuid)
- def __chainSizeCalc(self, sdUUID, imgUUID, volUUID, size):
+ def chainSizeCalc(self, sdUUID, imgUUID, volUUID, size):
"""
Compute an estimate of the whole chain size
using the sum of the actual size of the chain's volumes
@@ -763,7 +763,7 @@
if volParams['volFormat'] != volume.COW_FORMAT or \
volParams['prealloc'] != volume.SPARSE_VOL:
raise se.IncorrectFormat(self)
- volParams['apparentsize'] = self.__chainSizeCalc(
+ volParams['apparentsize'] = self.chainSizeCalc(
sdUUID, srcImgUUID, srcVolUUID, volParams['size'])
# Find out dest volume parameters
--
To view, visit https://gerrit.ovirt.org/38995
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Id079eb5067c16f934370e42b5f4e09bbcef1512b
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Adam Litke <alitke(a)redhat.com>