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@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
automation@ovirt.org has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 1:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
automation@ovirt.org has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 2:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
automation@ovirt.org has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 3:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
automation@ovirt.org has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 4:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
automation@ovirt.org has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 5:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
automation@ovirt.org has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 6:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
automation@ovirt.org has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 7:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
Roman Mohr has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 7:
@nir @francesco, adding you so that you are aware of it. Some initial feedback would be great when you find some time
Nir Soffer has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 7:
(14 comments)
https://gerrit.ovirt.org/#/c/47729/7/lib/vdsm/Makefile.am File lib/vdsm/Makefile.am:
Line 18: # Refer to the README and COPYING files for full details of the license Line 19: # Line 20: include $(top_srcdir)/build-aux/Makefile.subs Line 21: Line 22: SUBDIRS=netlink udev tool infra profiling We don't need udev package, udev module with Monitor class is fine and save lot of boilerplate. Line 23: Line 24: dist_vdsmpylib_PYTHON = \ Line 25: __init__.py \ Line 26: cmdutils.py \
https://gerrit.ovirt.org/#/c/47729/7/lib/vdsm/udev/__init__.py File lib/vdsm/udev/__init__.py:
Line 15: # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA Line 16: # Line 17: # Refer to the README and COPYING files for full details of the license Line 18: # Line 19: from __future__ import absolute_import The import is not needed, and empty files do not need a license.
https://gerrit.ovirt.org/#/c/47729/7/lib/vdsm/udev/monitor.py File lib/vdsm/udev/monitor.py:
Line 27: Line 28: """ Line 29: ``udev`` event monitor. Usage: Line 30: Line 31: The monitor is a thread-save thin wrapper arount pyudev.MonitorObserver. thread-safe Line 32: This allows multiple callbacks for the same netlink socket. To avoid Line 33: listening for udev events the application is not interested in, the Line 34: monitoring thread only starts listening on the socket when the monitor is Line 35: started and at least one subscriber is added.
Line 34: monitoring thread only starts listening on the socket when the monitor is Line 35: started and at least one subscriber is added. Line 36: Line 37: The simplest way to use the monitor is to subscribe a callback to a Line 38: specific subsystem event and let the callback do the work: This is called on the observer thread - right?
This will end in people blocking each other by invoking blocking calls during event handling .
It would be safer if we provide only a way to pull events from the monitor, instead of pushing the events to the subscribers.
I wonder if we should simply use multiple monitors instead of adding this wrapper - I guess it will be also more efficient.
This wrapper makes sense only if we have many monitors listening for the same events. But currently we don't have any monitors. I plan to monitor multipath devices regularly for updating lvm filter, and you wanted to monitor cpu events. So lets create 2 monitors and wait with this infrastructure until we need it. Line 39: Line 40: dev listen_for_disabled_cpu_events(device): Line 41: if device.action == 'offline': Line 42: print('CPU {0.name} is now offline'.format(device))
Line 60: monitor.start() Line 61: """ Line 62: Line 63: def __init__(self): Line 64: self._subsystems = {} This makes subscription handling little harder. Why not have a flat list of filters? Line 65: self._context = Context() Line 66: self._monitor = Monitor.from_netlink(self._context) Line 67: self._observer = MonitorObserver(self._monitor, callback=partial( Line 68: UdevMonitor._event_loop, self), name='udev-monitor')
Line 61: """ Line 62: Line 63: def __init__(self): Line 64: self._subsystems = {} Line 65: self._context = Context() Do we need to keep the context? Isn't it enough to create it and pass it to Monitor.from_netlink()? Line 66: self._monitor = Monitor.from_netlink(self._context) Line 67: self._observer = MonitorObserver(self._monitor, callback=partial( Line 68: UdevMonitor._event_loop, self), name='udev-monitor') Line 69: self._filter_lock = threading.Lock()
Line 64: self._subsystems = {} Line 65: self._context = Context() Line 66: self._monitor = Monitor.from_netlink(self._context) Line 67: self._observer = MonitorObserver(self._monitor, callback=partial( Line 68: UdevMonitor._event_loop, self), name='udev-monitor') Why do you need partial? Why not MonitorObserver(self._event_loop)? Line 69: self._filter_lock = threading.Lock() Line 70: self._is_started = False Line 71: self._can_start = False Line 72:
Line 69: self._filter_lock = threading.Lock() Line 70: self._is_started = False Line 71: self._can_start = False Line 72: Line 73: def _event_loop(self, device): This is not a good name, since this is not an event loop (it does not loop). Maybe _notify_subscribers? _handle_event? Line 74: subsystem = self._subsystems[device.subsystem] Line 75: for callback in subsystem.get(device.device_type, []): Line 76: _execute_callback(callback, device) Line 77: if device.device_type:
Line 77: if device.device_type: Line 78: for callback in subsystem.get(None, []): Line 79: _execute_callback(callback, device) Line 80: Line 81: def subscribe(self, callback, subsystem, device_type=None): If there is a way to subscribe, we need a way to unsubscribe. Line 82: """ Line 83: Raise :exc:`~exceptions.ValueError` if the callback is None Line 84: Line 85: :param callback: function to invoke
Line 92: """ Line 93: if callback is None: Line 94: raise ValueError('callback missing') Line 95: with self._filter_lock: Line 96: self._monitor.filter_by(subsystem, device_type) Does udev support removing filters? Line 97: device_types = self._subsystems.get(subsystem, {}) Line 98: callbacks = device_types.get(device_type, []) Line 99: callbacks.append(callback) Line 100: device_types[device_type] = callbacks
Line 97: device_types = self._subsystems.get(subsystem, {}) Line 98: callbacks = device_types.get(device_type, []) Line 99: callbacks.append(callback) Line 100: device_types[device_type] = callbacks Line 101: self._subsystems[subsystem] = device_types If we are using flat subscription list, this could be simplified to:
self._monitor.filter_by(subsystem, device_type) self._subscriptions[(subsystem, device_type)].append(callback) Line 102: self._start_if_necessary() Line 103: Line 104: def start(self): Line 105: self._can_start = True
Line 109: Line 110: def _start_if_necessary(self): Line 111: if self._can_start and not self._is_started: Line 112: self._observer.start() Line 113: self._is_started = True When unsubscribing the last filter, we can stop the observer. Line 114: Line 115: def stop(self): Line 116: """ Line 117: Stops the monitoring thread. It is guaranteed that callbacks are no
Line 128: try: Line 129: callback(device) Line 130: except Exception as callbackException: Line 131: logging.error( Line 132: 'Callback execution threw an exception: %s', callbackException) We like to see a traceback in this case, so this should be:
except Exception: logging.exception("Unhandled error in %s", callback)
https://gerrit.ovirt.org/#/c/47729/7/tests/Makefile.am File tests/Makefile.am:
Line 109: tcTests.py \ Line 110: testlibTests.py \ Line 111: toolTests.py \ Line 112: transportWrapperTests.py \ Line 113: udevMonitorTests.py \ If we move the monitor to udev.py, this should be udevTests.py Line 114: utilsTests.py \ Line 115: vdscliTests.py \ Line 116: vdsClientTests.py \ Line 117: vdsmDumpChainsTests.py \
Roman Mohr has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 7:
(4 comments)
https://gerrit.ovirt.org/#/c/47729/7/lib/vdsm/udev/monitor.py File lib/vdsm/udev/monitor.py:
Line 34: monitoring thread only starts listening on the socket when the monitor is Line 35: started and at least one subscriber is added. Line 36: Line 37: The simplest way to use the monitor is to subscribe a callback to a Line 38: specific subsystem event and let the callback do the work:
This is called on the observer thread - right? This will end in people blocking each other by invoking blocking calls during event handling.
Yes, it definitely can end up like that.
It would be safer if we provide only a way to pull events from the monitor, instead of pushing the events to the subscribers.
The idea is about having this as single point of source for non local sudden events. Events are handled one after the other (e.g. device is added and we have 10 listeners, then all that actions are executed before the next event comes. Would us allow to settle the new information throughout the application). Long running parts would be executed asynchronous or pushed to queues where a thread can poll, like you suggested.
This wrapper makes sense only if we have many monitors listening for the same events. But currently we don't have any monitors. I plan to monitor multipath devices regularly for updating lvm filter, and you wanted to monitor cpu events.
Valid objections. I have seen it as some kind of application wide service, where you only unsubscribe for shutdown. For actions of only local interest (like remove a device and wait until udev reports that it is really removed) someone can always just use the monitor from the pyudev library.
So lets create 2 monitors and wait with this infrastructure until we need it.
I am undecided. If you still insist after my comments, it is fine for me. Line 39: Line 40: dev listen_for_disabled_cpu_events(device): Line 41: if device.action == 'offline': Line 42: print('CPU {0.name} is now offline'.format(device))
Line 60: monitor.start() Line 61: """ Line 62: Line 63: def __init__(self): Line 64: self._subsystems = {}
This makes subscription handling little harder. Why not have a flat list of
Will do that if we decide to introduce the monitor. Line 65: self._context = Context() Line 66: self._monitor = Monitor.from_netlink(self._context) Line 67: self._observer = MonitorObserver(self._monitor, callback=partial( Line 68: UdevMonitor._event_loop, self), name='udev-monitor')
Line 77: if device.device_type: Line 78: for callback in subsystem.get(None, []): Line 79: _execute_callback(callback, device) Line 80: Line 81: def subscribe(self, callback, subsystem, device_type=None):
If there is a way to subscribe, we need a way to unsubscribe.
Unsubscribe is not possible because we can only remove all filters at once. Since the idea was to make this a service where you only unsubscribe for shutdown that, the general stop() method would be enough. Line 82: """ Line 83: Raise :exc:`~exceptions.ValueError` if the callback is None Line 84: Line 85: :param callback: function to invoke
Line 92: """ Line 93: if callback is None: Line 94: raise ValueError('callback missing') Line 95: with self._filter_lock: Line 96: self._monitor.filter_by(subsystem, device_type)
Does udev support removing filters?
You can only remove all at once Line 97: device_types = self._subsystems.get(subsystem, {}) Line 98: callbacks = device_types.get(device_type, []) Line 99: callbacks.append(callback) Line 100: device_types[device_type] = callbacks
Nir Soffer has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 7:
(2 comments)
https://gerrit.ovirt.org/#/c/47729/7/lib/vdsm/udev/monitor.py File lib/vdsm/udev/monitor.py:
Line 34: monitoring thread only starts listening on the socket when the monitor is Line 35: started and at least one subscriber is added. Line 36: Line 37: The simplest way to use the monitor is to subscribe a callback to a Line 38: specific subsystem event and let the callback do the work:
This will end in people blocking each other by invoking blocking calls du
Lets hear other people opinion on this. Line 39: Line 40: dev listen_for_disabled_cpu_events(device): Line 41: if device.action == 'offline': Line 42: print('CPU {0.name} is now offline'.format(device))
Line 77: if device.device_type: Line 78: for callback in subsystem.get(None, []): Line 79: _execute_callback(callback, device) Line 80: Line 81: def subscribe(self, callback, subsystem, device_type=None):
Unsubscribe is not possible because we can only remove all filters at once.
Not being able to unsubscribe is sad for publish and subscribe system. Line 82: """ Line 83: Raise :exc:`~exceptions.ValueError` if the callback is None Line 84: Line 85: :param callback: function to invoke
Jenkins CI RO has abandoned this change.
Change subject: monitor: Add udev monitor ......................................................................
Abandoned
Abandoned due to no activity - please restore if still relevant
gerrit-hooks has posted comments on this change.
Change subject: monitor: Add udev monitor ......................................................................
Patch Set 9:
* Update tracker: IGNORE, no Bug-Url found
vdsm-patches@lists.fedorahosted.org