Change in vdsm[master]: Added support for reporting the timezone and os info
by Vinzenz Feenstra
Vinzenz Feenstra has uploaded a new change for review.
Change subject: Added support for reporting the timezone and os info
......................................................................
Added support for reporting the timezone and os info
The guest agent reports now the timezone and more detailed os information.
This patch introduces the support for those new messages.
Change-Id: Iadb85a5003fcb4693b5a1174655acf59193340b5
Guest-Agent-API-Version: 2
Bug-Url: https://bugzilla.redhat.com/869296
Signed-off-by: Vinzenz Feenstra <vfeenstr(a)redhat.com>
---
M vdsm/rpc/vdsmapi-schema.json
M vdsm/virt/guestagent.py
2 files changed, 50 insertions(+), 2 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/40/28940/1
diff --git a/vdsm/rpc/vdsmapi-schema.json b/vdsm/rpc/vdsmapi-schema.json
index 2ccf963..869b78c 100644
--- a/vdsm/rpc/vdsmapi-schema.json
+++ b/vdsm/rpc/vdsmapi-schema.json
@@ -3456,6 +3456,12 @@
#
# @clientIp: The IP address of the client connected to the display
#
+# @guestOsInfo: #optional Guest OS information like architecture,
+# type, and version (Since 4.15.0)
+#
+# @guestTimezone: #optional Timezone and time offset of the guest OS
+# (Since 4.15.0)
+#
# Since: 4.14.1
##
{'type': 'VMFullInfo',
@@ -3470,7 +3476,45 @@
'guestFQDN': 'str', 'displayIp': 'str', 'keyboardLayout': 'str',
'displayPort': 'uint', 'guestIPs': 'str', 'smartcardEnable': 'bool',
'nicModel': 'VmInterfaceDeviceModel', 'pitReinjection': 'bool',
- 'status': 'str', 'clientIp': 'str'}}
+ 'status': 'str', 'clientIp': 'str', '*guestOsInfo': 'VMGuestOsInfo',
+ '*guestTimezone': 'VMGuestTimezone'}}
+
+##
+# @VMGuestTimezone:
+#
+# Timezone configuration of the guest operating system
+#
+# @zone: Name of the timezone native to the Guest OS
+#
+# @offset: Time offset of the configured timezone in minutes
+#
+# Since: 4.15.0
+##
+{'type': 'VMGuestTimezone',
+ 'data': {'zone': 'string', 'offset': 'uint'}}
+
+##
+# @VMGuestOsInfo:
+#
+# Information about the guest operating system on a VM
+#
+# @version: Operating system version
+#
+# @distribution: Name of the linux distribution
+#
+# @codename: Codename of the OS
+#
+# @arch: Architecture of the OS (x86, x86_64, ppc64, ...)
+#
+# @type: Type of the OS (linux, windows, ...)
+#
+# @kernel: Kernel version reported by a linux guest
+#
+# Since: 4.15.0
+##
+{'type': 'VMGuestOsInfo',
+ 'data': {'version': 'string', 'distribution': 'string', 'codename': 'string',
+ 'arch': 'string', 'type': 'string', 'kernel': 'string'}}
##
# @Host.getVMFullList:
diff --git a/vdsm/virt/guestagent.py b/vdsm/virt/guestagent.py
index 31c0a26..8ff262f 100644
--- a/vdsm/virt/guestagent.py
+++ b/vdsm/virt/guestagent.py
@@ -30,7 +30,7 @@
from . import vmstatus
-_MAX_SUPPORTED_API_VERSION = 1
+_MAX_SUPPORTED_API_VERSION = 2
_IMPLICIT_API_VERSION_ZERO = 0
_MESSAGE_API_VERSION_LOOKUP = {
@@ -255,6 +255,10 @@
self.guestInfo['guestName'] = args['name']
elif message == 'os-version':
self.guestInfo['guestOs'] = args['version']
+ elif message == 'os-info':
+ self.guestInfo['guestOsInfo'] = args
+ elif message == 'timezone':
+ self.guestInfo['guestTimezone'] = args
elif message == 'network-interfaces':
interfaces = []
old_ips = ''
--
To view, visit http://gerrit.ovirt.org/28940
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Iadb85a5003fcb4693b5a1174655acf59193340b5
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Vinzenz Feenstra <vfeenstr(a)redhat.com>
9 years, 4 months
Change in vdsm[master]: vm: Automatically add a serial port for a console device
by Vinzenz Feenstra
Vinzenz Feenstra has uploaded a new change for review.
Change subject: vm: Automatically add a serial port for a console device
......................................................................
vm: Automatically add a serial port for a console device
Without a serial device the console support does not work for
linux systems. We switch from virtio to serial and append a serial
port if a console device was defined.
Change-Id: Ifa7b02a7bcaad63017c35c811a194fa42e2b694f
Signed-off-by: Vinzenz Feenstra <vfeenstr(a)redhat.com>
---
M vdsm/vm.py
1 file changed, 14 insertions(+), 1 deletion(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/79/25979/1
diff --git a/vdsm/vm.py b/vdsm/vm.py
index 9cb0e82..7ce7c59 100644
--- a/vdsm/vm.py
+++ b/vdsm/vm.py
@@ -1171,6 +1171,18 @@
commandLine.appendChildWithArgs('qemu:arg', value='keyboard')
self.dom.appendChild(commandLine)
+ def appendSerial(self):
+ """
+ Add a serial port for the console device if it exists
+ <serial type='pty'>
+ <target port='0'>
+ </serial>
+ """
+ if len(self._devices.getElementsByTagName('console')) == 1:
+ s = XMLElement('serial', type='pty')
+ s.appendChildWithArgs('target', port='0')
+ self._devices.appendChild(s)
+
def appendGraphics(self):
"""
Add graphics section to domain xml.
@@ -1888,7 +1900,7 @@
</console>
"""
m = self.createXmlElem('console', 'pty')
- m.appendChildWithArgs('target', type='virtio', port='0')
+ m.appendChildWithArgs('target', type='serial', port='0')
return m
@@ -3025,6 +3037,7 @@
domxml.appendEmulator()
self._appendDevices(domxml)
+ domxml.appendSerial()
for drive in self._devices[DISK_DEVICES][:]:
for leaseElement in drive.getLeasesXML():
--
To view, visit http://gerrit.ovirt.org/25979
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ifa7b02a7bcaad63017c35c811a194fa42e2b694f
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Vinzenz Feenstra <vfeenstr(a)redhat.com>
9 years, 4 months
Change in vdsm[master]: net_tc: add support for mirroring all (not just IPv4) traffic
by asegurap@redhat.com
Antoni Segura Puimedon has uploaded a new change for review.
Change subject: net_tc: add support for mirroring all (not just IPv4) traffic
......................................................................
net_tc: add support for mirroring all (not just IPv4) traffic
Up until now the filter was only acting on packets with IPv4 headers
because of the 'protocol ip' part of the tc filter. This makes the
mirroring protocol agnostic.
Change-Id: I9de381e67fae4cb79fe41dbd9d1b60b72eb84de4
Signed-off-by: Antoni S. Puimedon <asegurap(a)redhat.com>
---
M vdsm/network/tc.py
1 file changed, 2 insertions(+), 2 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/88/27388/1
diff --git a/vdsm/network/tc.py b/vdsm/network/tc.py
index 35c4a92..de1837a 100644
--- a/vdsm/network/tc.py
+++ b/vdsm/network/tc.py
@@ -127,7 +127,7 @@
if filt.prio:
command.extend(['prio', filt.prio])
command.extend(['handle', filt.handle])
- command.extend(['protocol', 'ip', 'u32', 'match', 'u8', '0', '0'])
+ command.extend(['protocol', 'all', 'u32', 'match', 'u8', '0', '0'])
for a in filt.actions:
command.extend(['action', 'mirred', 'egress', 'mirror',
'dev', a.target])
@@ -203,7 +203,7 @@
out = _process_request([EXT_TC, 'filter', 'show', 'dev', dev,
'parent', parent])
- HEADER = 'filter protocol ip pref '
+ HEADER = 'filter protocol all pref '
prio = handle = None
actions = []
prevline = ' '
--
To view, visit http://gerrit.ovirt.org/27388
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I9de381e67fae4cb79fe41dbd9d1b60b72eb84de4
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Antoni Segura Puimedon <asegurap(a)redhat.com>
9 years, 4 months
Change in vdsm[master]: gluster: geo replication status and status detail
by dnarayan@redhat.com
Hello Bala.FA,
I'd like you to do a code review. Please visit
http://gerrit.ovirt.org/18414
to review the following change.
Change subject: gluster: geo replication status and status detail
......................................................................
gluster: geo replication status and status detail
this has two verbs, status: provides geo-replication status of all running
sessions or all sessions associated with a perticular source volume or
session between a source and remote volume. status detail: provides detailed
status of geo-repliction session between source and remote volume
Change-Id: I4f37f35a5480fbe049a67758e122d4a0c2eba513
Signed-off-by: ndarshan <dnarayan(a)redhat.com>
---
M client/vdsClientGluster.py
M vdsm/gluster/api.py
M vdsm/gluster/cli.py
M vdsm/gluster/exception.py
M vdsm/gluster/vdsmapi-gluster-schema.json
5 files changed, 223 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/14/18414/1
diff --git a/client/vdsClientGluster.py b/client/vdsClientGluster.py
index 90af83e..76a5ba8 100644
--- a/client/vdsClientGluster.py
+++ b/client/vdsClientGluster.py
@@ -424,6 +424,35 @@
pp.pprint(status)
return status['status']['code'], status['status']['message']
+ def do_glusterVolumeGeoRepStatus(self, args):
+ params = self._eqSplit(args)
+ try:
+ volName = params.get('volName', '')
+ remoteHost = params.get('remoteHost', '')
+ remoteVolName = params.get('remoteVolName', '')
+ except:
+ raise ValueError
+ status = self.s.glusterVolumeGeoRepStatus(volName, remoteHost,
+ remoteVolName)
+ pp.pprint(status)
+ return status['status']['code'], status['status']['message']
+
+ def do_glusterVolumeGeoRepStatusDetail(self, args):
+ params = self._eqSplit(args)
+ try:
+ volName = params.get('volName', '')
+ remoteHost = params.get('remoteHost', '')
+ remoteVolName = params.get('remoteVolName', '')
+ except:
+ raise ValueError
+ if not (volName and remoteHost and remoteVolName):
+ raise ValueError
+ status = self.s.glusterVolumeGeoRepStatusDetail(volName,
+ remoteHost,
+ remoteVolName)
+ pp.pprint(status)
+ return status['status']['code'], status['status']['message']
+
def getGlusterCmdDict(serv):
return \
@@ -705,4 +734,24 @@
'not set'
'(swift, glusterd, smb, memcached)'
)),
+ 'glusterVolumeGeoRepStatus': (
+ serv.do_glusterVolumeGeoRepStatus,
+ ('volName=<master_volume_name> '
+ 'remoteHost=<slave_host_name> '
+ 'remoteVolName=<slave_volume_name> '
+ '<master_volume_name>existing volume name in the master node\n\t'
+ '<slave_host_name>is remote slave host name or ip\n\t'
+ '<slave_volume_name>existing volume name in the slave node',
+ 'Returns ths status of geo-replication'
+ )),
+ 'glusterVolumeGeoRepStatusDetail': (
+ serv.do_glusterVolumeGeoRepStatusDetail,
+ ('volName=<master_volume_name> '
+ 'remoteHost=<slave_host_name> '
+ 'remoteVolName=<slave_volume_name> '
+ '<master_volume_name>existing volume name in the master node\n\t'
+ '<slave_host_name>is remote slave host name or ip\n\t'
+ '<slave_volume_name>existing volume name in the slave node',
+ 'Returns the Detailed status of geo-replication'
+ ))
}
diff --git a/vdsm/gluster/api.py b/vdsm/gluster/api.py
index 4bd8308..d24e700 100644
--- a/vdsm/gluster/api.py
+++ b/vdsm/gluster/api.py
@@ -287,6 +287,22 @@
status = self.svdsmProxy.glusterServicesGet(serviceNames)
return {'services': status}
+ @exportAsVerb
+ def volumeGeoRepStatus(self, volName=None, remoteHost=None,
+ remoteVolName=None, options=None):
+ status = self.svdsmProxy.glusterVolumeGeoRepStatus(volName,
+ remoteHost,
+ remoteVolName)
+ return {'geo-rep': status}
+
+ @exportAsVerb
+ def volumeGeoRepStatusDetail(self, volName, remoteHost,
+ remoteVolName, options=None):
+ status = self.svdsmProxy.glusterVolumeGeoRepStatusDetail(volName,
+ remoteHost,
+ remoteVolName)
+ return {'geo-rep': status}
+
def getGlusterMethods(gluster):
l = []
diff --git a/vdsm/gluster/cli.py b/vdsm/gluster/cli.py
index bac6d1c..1cf0e12 100644
--- a/vdsm/gluster/cli.py
+++ b/vdsm/gluster/cli.py
@@ -897,3 +897,59 @@
return _parseVolumeProfileInfo(xmltree, nfs)
except _etreeExceptions:
raise ge.GlusterXmlErrorException(err=[etree.tostring(xmltree)])
+
+
+def _parseGeoRepStatusDetail(tree):
+ status = {'node': tree.find('geoRep/node').text,
+ 'health': tree.find('geoRep/health').text,
+ 'uptime': tree.find('geoRep/uptime').text,
+ 'filesSyncd': tree.find('geoRep/filesSyncd').text,
+ 'filesPending': tree.find('geoRep/filesPending').text,
+ 'bytesPending': tree.find('geoRep/bytesPending').text,
+ 'deletesPending': tree.find('geoRep/deletesPending').text}
+ return status
+
+
+def _parseGeoRepStatus(tree):
+ pairs = []
+ for el in tree.findall('geoRep/pair'):
+ value = {}
+ value['node'] = el.find('node').text
+ value['master'] = el.find('master').text
+ value['slave'] = el.find('slave').text
+ value['health'] = el.find('health').text
+ value['uptime'] = el.find('uptime').text
+ pairs.append(value)
+ return pairs
+
+
+@makePublic
+def volumeGeoRepStatus(volName=None, remoteHost=None, remoteVolName=None,
+ ):
+ command = _getGlusterVolCmd() + ["geo-replication", volName,
+ "%s::%s" % (remoteHost, remoteVolName),
+ "status"]
+ try:
+ xmltree = _execGlusterXml(command)
+ except ge.GlusterCmdFailedException as e:
+ raise ge.GlusterGeoRepStatusFailedException(rc=e.rc, err=e.err)
+ try:
+ return _parseGeoRepStatus(xmltree)
+ except _etreeExceptions:
+ raise ge.GlusterXmlErrorException(err=[etree.tostring(xmltree)])
+
+
+@makePublic
+def volumeGeoRepStatusDetail(volName, remoteHost, remoteVolName,
+ ):
+ command = _getGlusterVolCmd() + ["geo-replication", volName,
+ "%s::%s" % (remoteHost, remoteVolName),
+ "status", "detail"]
+ try:
+ xmltree = _execGlusterXml(command)
+ except ge.GlusterCmdFailedException as e:
+ raise ge.GlusterGeoRepStatusDetailFailedException(rc=e.rc, err=e.err)
+ try:
+ return _parseGeoRepStatusDetail(xmltree)
+ except _etreeExceptions:
+ raise ge.GlusterXmlErrorException(err=[etree.tostring(xmltree)])
diff --git a/vdsm/gluster/exception.py b/vdsm/gluster/exception.py
index c569a9e..d95b168 100644
--- a/vdsm/gluster/exception.py
+++ b/vdsm/gluster/exception.py
@@ -484,3 +484,19 @@
prefix = "%s: " % (action)
self.message = prefix + "Service action is not supported"
self.err = [self.message]
+
+
+#geo-replication
+class GlusterGeoRepException(GlusterException):
+ code = 4560
+ message = "Gluster Geo-Replication Exception"
+
+
+class GlusterGeoRepStatusFailedException(GlusterGeoRepException):
+ code = 4565
+ message = "Geo Rep status failed"
+
+
+class GlusterGeoRepStatusDetailFailedException(GlusterGeoRepException):
+ code = 4566
+ message = "Geo Rep status detail failed"
diff --git a/vdsm/gluster/vdsmapi-gluster-schema.json b/vdsm/gluster/vdsmapi-gluster-schema.json
index 7a4c034..557c750 100644
--- a/vdsm/gluster/vdsmapi-gluster-schema.json
+++ b/vdsm/gluster/vdsmapi-gluster-schema.json
@@ -372,3 +372,89 @@
{'command': {'class': 'GlusterService', 'name': 'action'},
'data': {'serviceName': 'str', 'action': 'GlusterServiceAction'},
'returns': 'GlusterServicesStatusInfo'}
+
+##
+# @GlusterGeoRepStatus:
+#
+# Gluster geo-replication status information.
+#
+# @node: The node where geo-replication is started
+#
+# @master: The source for geo-replication
+#
+# @slave: The destination of geo-replication
+#
+# @health: The status of the geo-replication session
+#
+# @uptime: The time since the geo-replication started
+#
+# Since: 4.10.3
+##
+{'type': 'GlusterGeoRepStatus',
+ 'data': {'node': 'str', 'master': 'str', 'slave': 'str', 'health': 'str', 'uptime': 'int'}}
+
+
+##
+# @GlusterVolume.geoRepStatus:
+#
+# Gets the status of geo-Replication session
+#
+# @masterVolName: Is an existing volume name in the master node
+#
+# @slaveHost: Is remote slave host name or ip
+#
+# @slaveVolName: Is an available existing volume name in the slave node
+#
+# Returns:
+# status information for geo-replication
+#
+# Since: 4.10.3
+##
+{'command': {'class': 'GlusterVolume', 'name': 'geoRepStatus'},
+ 'data': {'masterVolName': 'str', 'slaveHost': 'str', 'slaveVolName': 'str'},
+ 'returns': 'GlusterGeoRepStatus'}
+
+##
+# @GlusterGeoRepStatusDetail:
+#
+# Gluster geo-replication detailed status information.
+#
+# @node: The node where geo-replication is started
+#
+# @health: The status of the geo-replication session
+#
+# @uptime: The time since the geo-replication started
+#
+# @filesSyncd: The number of files that are synced
+#
+# @filesPending: The number of files that are pending to be synced
+#
+# @bytesPending: The number of bytes that are pending to be synced
+#
+# @deletesPending: The number of deletes that are pending
+#
+# Since: 4.10.3
+##
+{'type': 'GlusterGeoRepStatusDetail',
+ 'data': {'node': 'str', 'health': 'str', 'uptime': 'int', 'filesSyncd': 'int', 'filesPending': 'int',
+ 'bytesPending': 'int','deletesPending': 'int'}}
+
+##
+# @GlusterVolume.geoRepStatusDetail:
+#
+# Gets the detailed status of geo-Replication session
+#
+# @masterVolName: Is an existing volume name in the master node
+#
+# @slaveHost: Is remote slave host name or ip
+#
+# @slaveVolName: Is an available existing volume name in the slave node
+#
+# Returns:
+# detailed status information of geo-replication session
+#
+# Since: 4.10.3
+##
+{'command': {'class': 'GlusterVolume', 'name': 'geoRepStatusDetail'},
+ 'data': {'masterVolName': 'str', 'slaveHost': 'str', 'slaveVolName': 'str'},
+ 'returns': 'GlusterGeoRepStatusDetail'}
--
To view, visit http://gerrit.ovirt.org/18414
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I4f37f35a5480fbe049a67758e122d4a0c2eba513
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: ndarshan <dnarayan(a)redhat.com>
Gerrit-Reviewer: Bala.FA <barumuga(a)redhat.com>
9 years, 4 months
Change in vdsm[master]: schedule: Introduce scheduling libary
by Nir Soffer
Nir Soffer has uploaded a new change for review.
Change subject: schedule: Introduce scheduling libary
......................................................................
schedule: Introduce scheduling libary
This moudule provides a Scheduler class scheduling execution of
callables on a background thread.
This should be part of the new scalable vm sampling implemntation, and
can be used also whenever you like to perform a short task on a
background thread, without waiting for the completion of the task.
See the module docstring and tests for usage examples.
Change-Id: Ie3764806d93bd37c3b5924080eb5ae4d29e4f4e0
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com>
---
M debian/vdsm-python.install
M lib/vdsm/Makefile.am
A lib/vdsm/schedule.py
M tests/Makefile.am
A tests/scheduleTests.py
M vdsm.spec.in
6 files changed, 346 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/07/29607/1
diff --git a/debian/vdsm-python.install b/debian/vdsm-python.install
index 2d4bba6..1775241 100644
--- a/debian/vdsm-python.install
+++ b/debian/vdsm-python.install
@@ -16,6 +16,7 @@
./usr/lib/python2.7/dist-packages/vdsm/netlink/link.py
./usr/lib/python2.7/dist-packages/vdsm/profile.py
./usr/lib/python2.7/dist-packages/vdsm/qemuimg.py
+./usr/lib/python2.7/dist-packages/vdsm/schedule.py
./usr/lib/python2.7/dist-packages/vdsm/sslutils.py
./usr/lib/python2.7/dist-packages/vdsm/tool/__init__.py
./usr/lib/python2.7/dist-packages/vdsm/tool/dummybr.py
diff --git a/lib/vdsm/Makefile.am b/lib/vdsm/Makefile.am
index c074bb3..89a5573 100644
--- a/lib/vdsm/Makefile.am
+++ b/lib/vdsm/Makefile.am
@@ -32,6 +32,7 @@
netinfo.py \
profile.py \
qemuimg.py \
+ schedule.py \
SecureXMLRPCServer.py \
sslutils.py \
utils.py \
diff --git a/lib/vdsm/schedule.py b/lib/vdsm/schedule.py
new file mode 100644
index 0000000..bd75924
--- /dev/null
+++ b/lib/vdsm/schedule.py
@@ -0,0 +1,213 @@
+#
+# Copyright 2014 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
+#
+
+"""
+This module provides a Scheduler class scheduling execution of
+a callable on a background thread.
+
+To use a scheduler, create an instance:
+
+ scheduler = schedule.Scheduler()
+
+When you want to schedule some callable:
+
+ def task():
+ print '30 seconds passed'
+
+ scheduler.schedule(30.0, task)
+
+task will be called after 30.0 seconds on the scheduler background thread.
+
+If you need to cancel a scheduled call, keep the ScheduledCall object returned
+from Scheduler.schedule(), and cancel the task:
+
+ scheduled_call = scheduler.schedule(30.0, call)
+ ...
+ scheduled_call.cancel()
+
+Finally, when the scheduler is not needed any more:
+
+ scheduler.cancel()
+
+This will cancel any pending calls and terminate the scheduler thread.
+"""
+
+import heapq
+import logging
+import threading
+import time
+
+from . import utils
+
+
+class Scheduler(object):
+ """
+ Schedule calls for future execution in a background thread.
+
+ This class is thread safe; multiple threads can schedule calls or cancel
+ the scheudler.
+ """
+
+ DEFAULT_DELAY = 30.0 # Used if no timeout are scheduled
+
+ _log = logging.getLogger("vds.Scheduler")
+
+ def __init__(self):
+ self._log.debug("Starting scheduler")
+ self._cond = threading.Condition(threading.Lock())
+ self._running = True
+ self._timeouts = []
+ t = threading.Thread(target=self._run)
+ t.daemon = True
+ t.start()
+
+ def schedule(self, delay, callee):
+ """
+ Schedule callee to be called after delay seconds on the scheduler
+ thread.
+
+ Callee must not block or take excessive time to complete. It it does
+ not finish quickly, it may delay other scheduled calls on the scheduler
+ thread.
+
+ Returns a ScheduledCall that may be canceled if callee was not called
+ yet.
+ """
+ deadline = time.time() + delay
+ timeout = _Timeout(deadline, callee)
+ self._log.debug("Schedulng %s", timeout)
+ with self._cond:
+ if self._running:
+ heapq.heappush(self._timeouts, timeout)
+ self._cond.notify()
+ else:
+ timeout.cancel()
+ return ScheduledCall(timeout)
+
+ def cancel(self):
+ """
+ Cancel all schedueld calls and invalidate the scheduler. Calls
+ scheduled after a scheduler was cancel will never be called.
+ """
+ self._log.debug("Canceling scheduler")
+ with self._cond:
+ self._running = False
+ self._cond.notify()
+
+ @utils.traceback(on=_log.name)
+ def _run(self):
+ try:
+ self._log.debug("started")
+ self._loop()
+ self._log.debug("canceled")
+ finally:
+ self._cleanup()
+
+ def _loop(self):
+ while True:
+ with self._cond:
+ if not self._running:
+ return
+ delay = self._time_until_deadline()
+ if delay > 0.0:
+ self._cond.wait(delay)
+ if not self._running:
+ return
+ expired = self._pop_expired_timeouts()
+ for timeout in expired:
+ timeout.fire()
+
+ def _time_until_deadline(self):
+ if self._timeouts:
+ return self._timeouts[0].deadline - time.time()
+ return self.DEFAULT_DELAY
+
+ def _pop_expired_timeouts(self):
+ now = time.time()
+ expired = []
+ while self._timeouts:
+ timeout = self._timeouts[0]
+ if timeout.deadline > now:
+ break
+ heapq.heappop(self._timeouts)
+ expired.append(timeout)
+ return expired
+
+ def _cleanup(self):
+ # Help the garbage collector by breaking reference cycles
+ with self._cond:
+ for timeout in self._timeouts:
+ timeout.cancel()
+
+
+class ScheduledCall(object):
+ """
+ Returned when a callable is scheduled to be called after delay. The caller
+ may cancel the call if it was not called yet.
+
+ This class is thread safe; any thread can cacnel a call.
+ """
+
+ _log = logging.getLogger("vds.Scheduler")
+
+ def __init__(self, timeout):
+ self._timeout = timeout
+
+ @property
+ def deadline(self):
+ return self._timeout.deadline
+
+ def cancel(self):
+ self._log.debug("Canceling %s", self)
+ self._timeout.cancel()
+
+
+# Sentinel for marking timeouts as invalid. Callable so we can invaliate a
+# timeout in a thread safe manner without locks.
+def _INVALID():
+ pass
+
+
+class _Timeout(object):
+ """
+ Created for each scheduled call.
+ """
+
+ _log = logging.getLogger('vds.Timeout')
+
+ def __init__(self, deadline, callee):
+ self.deadline = deadline
+ self.callee = callee
+
+ def fire(self):
+ if self.callee is _INVALID:
+ return
+ try:
+ self.callee()
+ except Exception:
+ self._log.exception("Unhandled exception in scheduled call")
+ finally:
+ self.callee = _INVALID
+
+ def cancel(self):
+ self.callee = _INVALID
+
+ def __cmp__(self, other):
+ return cmp(self.deadline, other.deadline)
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 6507165..786dea4 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -65,6 +65,7 @@
remoteFileHandlerTests.py \
resourceManagerTests.py \
samplingTests.py \
+ scheduleTests.py \
schemaTests.py \
securableTests.py \
sslTests.py \
diff --git a/tests/scheduleTests.py b/tests/scheduleTests.py
new file mode 100644
index 0000000..0c370d9
--- /dev/null
+++ b/tests/scheduleTests.py
@@ -0,0 +1,129 @@
+#
+# Copyright 2014 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
+#
+
+import threading
+import time
+
+from vdsm import schedule
+from testrunner import VdsmTestCase
+
+
+class SchedulerTests(VdsmTestCase):
+
+ # Time to wait for completion, so test will not fail on overloaded
+ # machines. If tests fails on CI, increase this value.
+ GRACETIME = 0.1
+
+ MAX_TASKS = 1000
+
+ def setUp(self):
+ self.scheduler = schedule.Scheduler()
+
+ def tearDown(self):
+ self.scheduler.cancel()
+
+ def test_schedule(self):
+ delay = 0.3
+ task1 = Task()
+ task2 = Task()
+ timeout1 = self.scheduler.schedule(delay, task1)
+ self.scheduler.schedule(10, task2)
+ task1.wait(delay + self.GRACETIME)
+ self.assertTrue(timeout1.deadline <= task1.call_time)
+ self.assertTrue(task1.call_time < timeout1.deadline + self.GRACETIME)
+ self.assertEquals(task2.call_time, None)
+
+ def test_schedule_many(self):
+ delay = 0.3
+ tasks = []
+ for i in range(self.MAX_TASKS):
+ task = Task()
+ timeout = self.scheduler.schedule(delay, task)
+ tasks.append((task, timeout))
+ last_task = tasks[-1][0]
+ last_task.wait(delay + self.GRACETIME)
+ for task, timeout in tasks:
+ self.assertTrue(timeout.deadline <= task.call_time)
+ self.assertTrue(task.call_time < timeout.deadline + self.GRACETIME)
+
+ def test_continue_after_failures(self):
+ self.scheduler.schedule(0.3, FailingTask())
+ task = Task()
+ self.scheduler.schedule(0.4, task)
+ task.wait(0.4 + self.GRACETIME)
+ self.assertTrue(task.call_time is not None)
+
+ def test_cancel_timeout(self):
+ delay = 0.3
+ task = Task()
+ timeout = self.scheduler.schedule(delay, task)
+ timeout.cancel()
+ task.wait(delay + self.GRACETIME)
+ self.assertEquals(task.call_time, None)
+
+ def test_cancel_many(self):
+ delay = 0.3
+ tasks = []
+ for i in range(self.MAX_TASKS):
+ task = Task()
+ timeout = self.scheduler.schedule(delay, task)
+ tasks.append((task, timeout))
+ for task, timeout in tasks:
+ timeout.cancel()
+ last_task = tasks[-1][0]
+ last_task.wait(delay + self.GRACETIME)
+ for task, timeout in tasks:
+ self.assertEquals(task.call_time, None)
+
+ def test_cancel(self):
+ delay = 0.3
+ tasks = []
+ for i in range(self.MAX_TASKS):
+ task = Task()
+ timeout = self.scheduler.schedule(delay, task)
+ tasks.append((task, timeout))
+ self.scheduler.cancel()
+ last_task = tasks[-1][0]
+ last_task.wait(delay + self.GRACETIME)
+ for task, timeout in tasks:
+ self.assertEquals(task.call_time, None)
+
+
+class Task(object):
+
+ def __init__(self):
+ self.cond = threading.Condition(threading.Lock())
+ self.call_time = None
+
+ def __call__(self):
+ with self.cond:
+ self.call_time = time.time()
+ self.cond.notify()
+
+ def wait(self, timeout):
+ with self.cond:
+ if self.call_time is None:
+ self.cond.wait(timeout)
+
+
+class FailingTask(object):
+
+ def __call__(self):
+ raise Exception("This task is broken")
diff --git a/vdsm.spec.in b/vdsm.spec.in
index dfca5bd..8ba0477 100644
--- a/vdsm.spec.in
+++ b/vdsm.spec.in
@@ -1163,6 +1163,7 @@
%{python_sitearch}/%{vdsm_name}/qemuimg.py*
%{python_sitearch}/%{vdsm_name}/SecureXMLRPCServer.py*
%{python_sitearch}/%{vdsm_name}/netconfpersistence.py*
+%{python_sitearch}/%{vdsm_name}/schedule.py*
%{python_sitearch}/%{vdsm_name}/sslutils.py*
%{python_sitearch}/%{vdsm_name}/utils.py*
%{python_sitearch}/%{vdsm_name}/vdscli.py*
--
To view, visit http://gerrit.ovirt.org/29607
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ie3764806d93bd37c3b5924080eb5ae4d29e4f4e0
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com>
9 years, 4 months
Change in vdsm[master]: threadpool: add a thread pool with watchdog
by fromani@redhat.com
Francesco Romani has uploaded a new change for review.
Change subject: threadpool: add a thread pool with watchdog
......................................................................
threadpool: add a thread pool with watchdog
Add a new thread pool implementation with built-in
watchdog.
The watchdog is needed to accomodate the needs of
the virt sampling code.
The virt sampling needs to deal with possibly blocking
I/O like calls which cannot be made not-blocking.
Change-Id: Ic06da1ba57868dc2c7db67a1868ad10087a1cff2
Signed-off-by: Francesco Romani <fromani(a)redhat.com>
---
M lib/threadpool/Makefile.am
A lib/threadpool/examples/demo.py
A lib/threadpool/schedqueue.py
A lib/threadpool/tests/test_schedqueue.py
A lib/threadpool/tests/test_timetrackingthread.py
A lib/threadpool/tests/test_worker.py
A lib/threadpool/tests/test_workqueue.py
A lib/threadpool/watchedpool.py
A lib/threadpool/watchman.py
A lib/threadpool/worker.py
A lib/threadpool/workqueue.py
11 files changed, 1,003 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/91/29191/1
diff --git a/lib/threadpool/Makefile.am b/lib/threadpool/Makefile.am
index 3a71a5d..8b7098e 100644
--- a/lib/threadpool/Makefile.am
+++ b/lib/threadpool/Makefile.am
@@ -22,4 +22,9 @@
dist_threadpool_PYTHON = \
__init__.py \
+ schedqueue.py \
+ watchedpool.py \
+ watchman.py \
+ worker.py \
+ workqueue.py \
$(NULL)
diff --git a/lib/threadpool/examples/demo.py b/lib/threadpool/examples/demo.py
new file mode 100644
index 0000000..cf80930
--- /dev/null
+++ b/lib/threadpool/examples/demo.py
@@ -0,0 +1,62 @@
+#
+# Copyright 2014 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
+#
+
+import logging
+import random
+import sys
+import time
+
+from watchedpool import WatchedThreadPool
+
+
+def _test_fun(*args):
+ """
+ just block the caller for a random time span.
+ """
+ tmo = random.randint(1, 15)
+ logging.info('started, waiting: %.3fs', tmo)
+ time.sleep(tmo)
+ logging.info('done')
+
+
+def _main(args):
+ """
+ main test driver
+ """
+ nworkers = int(args[0]) if len(args) >= 1 else 1
+ pool = WatchedThreadPool(nworkers)
+ pool.start()
+ for tag in range(nworkers):
+ pool.submit_periodic(4, _test_fun)
+
+ try:
+ while True:
+ time.sleep(1.0)
+ except KeyboardInterrupt:
+ logging.debug('exiting!')
+ finally:
+ pool.stop()
+
+
+if __name__ == '__main__':
+ logging.basicConfig(
+ level=logging.DEBUG,
+ format='%(asctime)s %(threadName)-10s %(levelname)-8s %(message)s')
+ _main(sys.argv[1:])
diff --git a/lib/threadpool/schedqueue.py b/lib/threadpool/schedqueue.py
new file mode 100644
index 0000000..c2bfbae
--- /dev/null
+++ b/lib/threadpool/schedqueue.py
@@ -0,0 +1,123 @@
+#
+# Copyright 2014 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
+#
+
+"""
+Time-aware priority queue, built like queue.Queue
+but tailored for WatchedThreadPool needs.
+"""
+
+import heapq
+import threading
+import time
+
+
+class QueueError(Exception):
+ """
+ Queue exception root.
+ """
+
+
+class Empty(QueueError):
+ """
+ Attempt to get an item from an empty queue.
+ """
+
+
+class Full(QueueError):
+ """
+ Attempt to put an item into a full queue.
+ """
+
+
+class NotYet(QueueError):
+ """
+ Attempt to get an item which is not due at the
+ present (reported) time.
+ """
+
+
+class SchedQueue(object):
+ """
+ Time-aware, not blocking (client must take care of
+ waiting/timing) priority queue.
+ """
+ def __init__(self, maxsize=0, timefunc=time.time):
+ self._timefunc = timefunc
+ self._maxsize = maxsize
+ self._queue = self._init(maxsize)
+ self._lock = threading.Lock()
+
+ def __len__(self):
+ with self._lock:
+ return self._size()
+
+ def put(self, item, delay=0):
+ """
+ Put an item into the queue, to be made ready 'delay' seconds
+ in the future.
+ """
+ with self._lock:
+ if self._maxsize > 0 and self._size() == self._maxsize:
+ raise Full
+ self._put((self._timefunc() + delay, item))
+
+ def get(self, timenow):
+ """
+ Ask for an item to be consumed.
+ """
+ with self._lock:
+ if not self._size():
+ raise Empty
+ timeval, _ = self._peek()
+ if timeval > timenow:
+ raise NotYet
+ _, item = self._get()
+ return item
+
+ def _init(self, maxsize):
+ """
+ builds the internal queue.
+ """
+ self._maxsize = maxsize # to make pylint happy
+ return []
+
+ def _size(self):
+ """
+ size of the internal queue.
+ """
+ return len(self._queue)
+
+ def _put(self, item):
+ """
+ put a new element (at the end of) in the queue.
+ """
+ heapq.heappush(self._queue, item)
+
+ def _get(self):
+ """
+ gets (and removes) the next element in queue.
+ """
+ return heapq.heappop(self._queue)
+
+ def _peek(self):
+ """
+ read-only view of the next element in queue.
+ """
+ return self._queue[0]
diff --git a/lib/threadpool/tests/test_schedqueue.py b/lib/threadpool/tests/test_schedqueue.py
new file mode 100644
index 0000000..3595589
--- /dev/null
+++ b/lib/threadpool/tests/test_schedqueue.py
@@ -0,0 +1,56 @@
+#
+# Copyright 2014 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
+#
+
+import unittest
+
+import schedqueue
+
+
+class SchedQueueTests(unittest.TestCase):
+ def _faketime(self):
+ ret = self._ts
+ self._ts += 1
+ return ret
+
+ def setUp(self):
+ self._ts = 0
+ self._sq = schedqueue.SchedQueue(timefunc=self._faketime)
+
+ def test_empty_on_create(self):
+ self.assertEqual(len(self._sq), 0)
+
+ def test_put_on_full(self):
+ self._sq = schedqueue.SchedQueue(maxsize=1, timefunc=self._faketime)
+ self._sq.put('foo')
+ self.assertRaises(schedqueue.Full, self._sq.put, 'bar')
+
+ def test_get_on_empty(self):
+ self.assertRaises(schedqueue.Empty, self._sq.get, 1)
+
+ def test_get_too_early(self):
+ item = 'foo'
+ self._sq.put(item, delay=5)
+ self.assertRaises(schedqueue.NotYet, self._sq.get, 0)
+
+ def test_put_get(self):
+ item = 'foo'
+ self._sq.put(item)
+ res = self._sq.get(timenow=1)
+ self.assertEqual(res, item)
diff --git a/lib/threadpool/tests/test_timetrackingthread.py b/lib/threadpool/tests/test_timetrackingthread.py
new file mode 100644
index 0000000..14a25ef
--- /dev/null
+++ b/lib/threadpool/tests/test_timetrackingthread.py
@@ -0,0 +1,64 @@
+#
+# Copyright 2014 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
+#
+
+import unittest
+
+import worker
+
+
+class TimeTrackingThreadTests(unittest.TestCase):
+ def _fakesleep(self, delay):
+ self._ts += max(0, delay) # avoid negative delays
+
+ def _faketime(self):
+ ret = self._ts
+ self._ts += 1
+ return ret
+
+ def setUp(self):
+ self._ts = 0
+ self._ttt = worker.TimeTrackingThread(self._faketime)
+
+ def assertIdle(self):
+ self.assertFalse(self._ttt.busy)
+ self.assertEqual(self._ttt.task, None)
+
+ def assertBusyOn(self, task):
+ self.assertTrue(self._ttt.busy)
+ self.assertEqual(self._ttt.task, task)
+
+ def test_created_free(self):
+ self.assertIdle()
+
+ def test_carry_task(self):
+ self.assertIdle()
+
+ task = 'foobar'
+ with self._ttt.track_time(task):
+ self.assertBusyOn(task)
+
+ self.assertIdle()
+
+ def test_elapsed(self):
+ task = 'barbaz'
+ delay = 5
+ with self._ttt.track_time(task):
+ self._fakesleep(delay)
+ self.assertTrue(self._ttt.elapsed >= delay)
diff --git a/lib/threadpool/tests/test_worker.py b/lib/threadpool/tests/test_worker.py
new file mode 100644
index 0000000..7396673
--- /dev/null
+++ b/lib/threadpool/tests/test_worker.py
@@ -0,0 +1,60 @@
+#
+# Copyright 2014 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
+#
+
+import contextlib
+import unittest
+
+import schedqueue
+import worker
+
+
+class FakeWorkQueue(object):
+ def __init__(self, items):
+ self._items = items
+
+ @contextlib.contextmanager
+ def fetch(self):
+ if not self._items:
+ raise schedqueue.Empty
+ item = self._items.pop(0)
+ if issubclass(item, Exception):
+ raise item
+ else:
+ yield item
+
+
+class WorkerThreadTests(unittest.TestCase):
+ def test__pull_empty(self):
+ wq = FakeWorkQueue([])
+ wt = worker.Worker(wq)
+ with wt._pull() as item:
+ self.assertEqual(item, (None, None, None, None))
+
+ def test__pull_empty_exc(self):
+ wq = FakeWorkQueue([schedqueue.Empty])
+ wt = worker.Worker(wq)
+ with wt._pull() as item:
+ self.assertEqual(item, (None, None, None, None))
+
+ def test__pull_notyet(self):
+ wq = FakeWorkQueue([schedqueue.NotYet])
+ wt = worker.Worker(wq)
+ with wt._pull() as item:
+ self.assertEqual(item, (None, None, None, None))
diff --git a/lib/threadpool/tests/test_workqueue.py b/lib/threadpool/tests/test_workqueue.py
new file mode 100644
index 0000000..68e982d
--- /dev/null
+++ b/lib/threadpool/tests/test_workqueue.py
@@ -0,0 +1,80 @@
+#
+# Copyright 2014 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
+#
+
+import unittest
+
+import schedqueue
+import watchedpool
+
+
+def _do_nothing(*args, **kwargs):
+ pass
+
+
+class WorkQueueTests(unittest.TestCase):
+ def _fakesleep(self, delay):
+ pass
+
+ def _faketime(self):
+ ret = self._ts
+ self._ts += 1
+ return ret
+
+ def assertDummyWork(self, workitem, in_tag=None):
+ out_tag, work, args, kwargs = workitem
+ if in_tag is not None:
+ self.assertEqual(in_tag, out_tag)
+ self.assertEqual(work, _do_nothing)
+ self.assertEqual(len(args), 0)
+ self.assertEqual(len(kwargs), 0)
+
+ def setUp(self):
+ self._ts = 0
+ self._wq = watchedpool.WorkQueue(
+ timefunc=self._faketime,
+ delayfunc=self._fakesleep)
+
+ def test_created_empty(self):
+ self.assertEqual(len(self._wq), 0)
+
+ def test_get_on_empty(self):
+ self.assertRaises(schedqueue.Empty, self._wq.get)
+
+ def test_post_on_empty(self):
+ self._wq.post(0, _do_nothing)
+ self.assertEqual(len(self._wq), 1)
+
+ def test_put_get(self):
+ in_tag = 'foo'
+ self._wq.put(in_tag, 0, _do_nothing)
+ self.assertDummyWork(self._wq.get())
+
+ def test_remove(self):
+ in_tag = 'foo'
+ self._wq.put(in_tag, 0, _do_nothing)
+ self._wq.remove(in_tag)
+ self.assertRaises(schedqueue.Empty, self._wq.get)
+
+ def test_fetch(self):
+ in_tag = 'foo'
+ self._wq.put(in_tag, 1, _do_nothing)
+ with self._wq.fetch() as work_item:
+ self.assertDummyWork(work_item)
+ self.assertEqual(len(self._wq), 1)
diff --git a/lib/threadpool/watchedpool.py b/lib/threadpool/watchedpool.py
new file mode 100644
index 0000000..6a8938f
--- /dev/null
+++ b/lib/threadpool/watchedpool.py
@@ -0,0 +1,128 @@
+#
+# Copyright 2014 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
+#
+
+"""
+PoC of a watched thread pool implementation.
+This thread pool adds a watcher/watchman thread which detaches `rogue'
+threads from the main pool, and ask them to die as soon as possible.
+The purpose is to avoid worker threads getting stuck too much on
+busy, possible blocking operations (e.g. risky I/O).
+
+In the worst case scenario, the pool size degenerates in a
+thread-per-operation. In the common and expected case,
+the pool size is fixed and rogue threads are just spikes
+in the thread number.
+
+TODO:
+- allow the client to resize dynamically the pool [?]
+- let this code be easily mergeable with vdsm/storage/threadPool.py
+- demonstrate that this code does not leak threads.
+"""
+
+import logging
+import threading
+
+from watchman import Watchman
+from worker import Worker
+from workqueue import WorkQueue
+
+
+class WatchedThreadPool(object):
+ """
+ Straightforward thread pool, with an added Watchman.
+ """
+ def __init__(self, nworkers=1, work_interval=1):
+ self._nworkers = nworkers
+ self._work_interval = work_interval
+ self._watch_interval = 2 * self._work_interval
+ self._busy_timeout = 3 * self._watch_interval
+ self._work_queue = None
+ self._workers = None
+ self._watcher = None
+ self._running = False
+ self._lock = threading.Lock()
+
+ def start(self):
+ """
+ starts the pool. Thread safe.
+ """
+ with self._lock:
+ if not self._running:
+ self._work_queue = WorkQueue(self._work_interval)
+ self._workers = [self._make_worker()
+ for _ in range(self._nworkers)]
+ self._watcher = Watchman(self._workers,
+ self._make_worker,
+ self._watch_interval,
+ self._busy_timeout)
+ self._watcher.start()
+ logging.info('pool started')
+ self._running = True
+
+ def stop(self, wait=True):
+ """
+ terminate all pooled threads. Thread safe.
+ """
+ with self._lock:
+ if self._running:
+ self._running = False
+
+ self._watcher.stop()
+ if wait:
+ for worker in self._workers:
+ worker.join()
+ del worker
+ self._watcher.reap([]) # last chance
+ self._watcher.join()
+ del self._watcher
+
+ def submit_periodic(self, interval, work, *args, **kwargs):
+ """
+ submit a new work unit to the pool.
+ returns the work unit tag.
+ The tag is guaranteed to be unique per-pool.
+ """
+ if self._running:
+ tag = self._work_queue.post(interval, work, args, kwargs)
+ logging.debug('new work: (%s, %s, %s) -> %s',
+ str(work), str(args), str(kwargs), tag)
+ return tag # FIXME
+ else:
+ return None
+
+ def remove(self, tag):
+ """
+ remove a work unit from the pool.
+ """
+ return self._work_queue.remove(tag)
+
+ def notify_timeout(self, tag, on_timeout):
+ """
+ register a timeout callback for the given work unit.
+ """
+ self._watcher.notify_timeout(tag, on_timeout)
+
+ def _make_worker(self):
+ """
+ creates a new idle (not started) worker.
+ """
+ worker = Worker(self._work_queue)
+ logging.info('worker %s added to the pool', worker.name)
+ return worker
diff --git a/lib/threadpool/watchman.py b/lib/threadpool/watchman.py
new file mode 100644
index 0000000..53bfc11
--- /dev/null
+++ b/lib/threadpool/watchman.py
@@ -0,0 +1,209 @@
+#
+# Copyright 2014 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
+#
+
+"""
+PoF of a watched thread pool implementation.
+This thread pool adds a watcher/watchman thread which detaches `rogue'
+threads from the main pool, and ask them to die as soon as possible.
+The purpose is to avoid worker threads getting stuck too much on
+busy, possible blocking operations (e.g. risky I/O).
+
+In the worst case scenario, the pool size degenerates in a
+thread-per-operation. In the common and expected case,
+the pool size is fixed and rogue threads are just spikes
+in the thread number.
+
+TODO:
+- allow the client to resize dynamically the pool [?]
+- let this code be easily mergeable with vdsm/storage/threadPool.py
+- demonstrate that this code does not leak threads.
+"""
+
+import logging
+import threading
+import time
+
+
+class Watchman(threading.Thread):
+ """
+ Watches the worker threads, detects the rogues, if any,
+ and possibly replace them to ensure the pool is operative
+ and not blocked.
+ """
+ def __init__(self, workers, make_worker,
+ watch_interval=5, busy_timeout=10,
+ rogues_cap=None, delayfunc=time.sleep):
+ super(Watchman, self).__init__()
+ self.daemon = True
+ self._workers = workers
+ self._rogues = []
+ self._rogues_cap = rogues_cap
+ self._watch_interval = watch_interval
+ self._make_worker = make_worker
+ self._busy_timeout = busy_timeout
+ self._timeout_cb = {}
+ self._stop = threading.Event()
+ self._running = True
+ self._delayfunc = delayfunc
+
+ def suspend(self):
+ """
+ suspend the watch, without stopping.
+ """
+ self._running = False
+
+ def resume(self):
+ """
+ resume the watch.
+ """
+ self._running = True
+
+ def start(self):
+ """
+ start the watched pool and the watcher itself.
+ """
+ for worker in self._workers:
+ worker.start()
+ super(Watchman, self).start()
+
+ def stop(self):
+ """
+ stop the watched pool, and the watcher itself.
+ """
+ self.suspend()
+ for worker in self._workers:
+ worker.stop()
+ self._stop.set()
+
+ def run(self):
+ logging.debug(
+ 'watcher starting (with %i workers)', len(self._workers))
+
+ while not self._stop.is_set():
+ self._delayfunc(self._watch_interval)
+ if self._running:
+ workers, rogues, newbies = self.verify()
+ self.enforce(workers, rogues, newbies)
+
+ logging.debug('watcher done')
+
+ def notify_timeout(self, tag, on_timeout):
+ """
+ register a timeout callback for the given work unit.
+ """
+ self._timeout_cb[tag] = on_timeout
+
+ def stuck(self, worker):
+ """
+ is a given worker stuck?
+ """
+ stuck, task = False, None
+ if worker.busy:
+ # use locals to freeze the state
+ task = worker.task
+ elapsed = worker.elapsed
+ stuck = elapsed > self._busy_timeout
+ logging.warning(
+ '%s busy on %s for the last %02i seconds (max=%02i) -> %s',
+ worker.name, task, elapsed, self._busy_timeout,
+ 'STUCK' if stuck else 'OK')
+ return stuck, task
+
+ def verify(self):
+ """
+ check the health of the worker pool, and classify
+ it into
+ - workers: well behaving worker threads
+ - rogues: workers got stuck, to be evicted from the pool
+ - newbies: new workers ready to replace rogues
+ """
+ try:
+ workers, rogues, newbies = [], [], []
+ logging.debug('verification loop started')
+ for worker in self._workers:
+ stuck, task = self.stuck(worker)
+ if not stuck:
+ workers.append(worker)
+ elif not self._room_for_rogues(rogues):
+ logging.warning(
+ 'worker %s gone rogue, but cap reached', worker.name)
+ elif not self._is_detachable(task):
+ logging.warning(
+ 'worker %s gone rogue, but task %s not detachable',
+ worker.name, task)
+ else:
+ logging.info(
+ 'worker %s gone rogue, replacing', worker.name)
+ worker.stop()
+ rogues.append(worker) # until it dies
+ newbies.append(self._make_worker())
+ logging.debug('verification loop done: %i rogues', len(rogues))
+ return workers, rogues, newbies
+ except Exception:
+ logging.exception('verify failed', exc_info=True)
+
+ def enforce(self, workers, rogues, newbies):
+ """
+ enforce the pool size by evicting rogues into a limbo,
+ and by replenishing the worker pool with the provided
+ newbies.
+ """
+ try:
+ for newbie in newbies:
+ newbie.start()
+ workers.append(newbie)
+ self._workers = workers
+ self.reap(rogues)
+ except Exception:
+ logging.exception('enforce failed', exc_info=True)
+
+ def reap(self, rogues):
+ """
+ collect the terminated rogues.
+ """
+ rogues.extend(self._rogues)
+ new_rogues = []
+ for rogue in rogues:
+ if rogue.is_alive():
+ new_rogues.append(rogue)
+ else:
+ logging.info('rogue thread %s collected', rogue.name)
+ rogue.join()
+ del rogue
+ self._rogues = new_rogues
+
+ def _room_for_rogues(self, rogues):
+ return (self._rogues_cap is None
+ or len(rogues) < self._rogues_cap)
+
+ def _is_detachable(self, task):
+ """
+ if a task is detachable, its worker thread will be detached
+ by the pool. Otherwise the thread will be kept in the pool
+ """
+ on_timeout = self._timeout_cb.get(task, _on_timeout_dummy)
+ return on_timeout(task)
+
+
+def _on_timeout_dummy(*args):
+ """
+ Dummy timeout callback. Marks every work unit as detachable.
+ """
+ return True
diff --git a/lib/threadpool/worker.py b/lib/threadpool/worker.py
new file mode 100644
index 0000000..fe63d34
--- /dev/null
+++ b/lib/threadpool/worker.py
@@ -0,0 +1,129 @@
+#
+# Copyright 2014 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
+#
+
+import contextlib
+import logging
+import threading
+import time
+
+import schedqueue
+
+
+class TimeTrackingThread(threading.Thread):
+ """
+ Thin mixin to help a worker thread to track the time
+ consumed by a work item.
+ """
+ def __init__(self, timefunc=time.time):
+ super(TimeTrackingThread, self).__init__()
+ self._work_started_at = None
+ self._working_on = None
+ self._timefunc = timefunc
+
+ @property
+ def busy(self):
+ """
+ is the thread doing some work or is it idle?
+ """
+ return self._work_started_at is not None
+
+ @property
+ def elapsed(self):
+ """
+ how much time since the thread started the current work unit?
+ """
+ return self._timefunc() - self._work_started_at
+
+ @property
+ def task(self):
+ """
+ what is this worker doing?
+ """
+ return self._working_on
+
+ @contextlib.contextmanager
+ def track_time(self, tag):
+ """
+ helper context manager to track the time consumed in
+ the current work unit
+ """
+ self._working_on = tag
+ self._work_started_at = self._timefunc()
+ yield
+ self._work_started_at = None
+ self._working_on = None
+
+
+class Worker(TimeTrackingThread):
+ """
+ A regular worker thread. This pool has no additional
+ requirements saved the ability of a worker thread
+ to die ASAP (see the stop() method)
+ """
+ def __init__(self, work_queue):
+ super(Worker, self).__init__()
+ self.daemon = True
+ self._work_queue = work_queue
+ self._stop = threading.Event()
+
+ def run(self):
+ logging.info('worker %s starting', self.name)
+
+ while not self._stop.is_set():
+ self.process()
+
+ logging.info('worker %s done', self.name)
+
+ def process(self):
+ """
+ execute a single try of work processing.
+ """
+ with self._pull() as (tag, work, args, kwargs):
+ if work is not None:
+ self._do_work(tag, work, args, kwargs)
+
+ def stop(self):
+ """
+ gracefully terminate ASAP, and avoid to begin more work.
+ """
+ self._stop.set()
+
+ @contextlib.contextmanager
+ def _pull(self):
+ """
+ pulls the next work item
+ """
+ try:
+ with self._work_queue.fetch() as item:
+ yield item
+ except schedqueue.NotYet:
+ yield None, None, None, None
+ except schedqueue.Empty:
+ yield None, None, None, None
+
+ def _do_work(self, tag, work, args, kwargs):
+ """
+ process a single work unit.
+ """
+ with self.track_time(tag):
+ try:
+ work(*args, **kwargs)
+ except Exception:
+ logging.exception('work failed', exc_info=True)
diff --git a/lib/threadpool/workqueue.py b/lib/threadpool/workqueue.py
new file mode 100644
index 0000000..73949b8
--- /dev/null
+++ b/lib/threadpool/workqueue.py
@@ -0,0 +1,87 @@
+#
+# Copyright 2014 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
+#
+
+import contextlib
+import time
+import uuid
+
+import schedqueue
+
+
+class WorkQueue(object):
+ """
+ Tiny wrapper around a plain Queue.Queue, to encapsulate
+ the 'periodic work' behaviour.
+ """
+ def __init__(self, wait_timeout=1,
+ timefunc=time.time, delayfunc=time.sleep):
+ self._queue = schedqueue.SchedQueue(timefunc=timefunc)
+ self._work = {}
+ self._wait_timeout = wait_timeout
+ self._timefunc = timefunc
+ self._delayfunc = delayfunc
+
+ def put(self, tag, interval, work, *args, **kwargs):
+ """
+ like Queue.put().
+ """
+ item = (work, args, kwargs, interval)
+ self._work[tag] = item
+ self._queue.put(tag, delay=interval)
+
+ def get(self):
+ """
+ like Queue.get().
+ """
+ tag, work, args, kwargs, _ = self._get()
+ return (tag, work, args, kwargs)
+
+ def remove(self, tag):
+ return self._work.pop(tag, None)
+
+ def post(self, interval, work, *args, **kwargs):
+ """
+ like put(), but with auto-tagging.
+ """
+ return self.put(str(uuid.uuid4()), interval, work, *args, **kwargs)
+
+ @contextlib.contextmanager
+ def fetch(self):
+ """
+ like get(), but automatically reinject the work
+ to do in the queue if the task is periodic.
+ """
+ tag, work, args, kwargs, interval = self._get()
+ yield (tag, work, args, kwargs)
+ if interval:
+ self.put(tag, interval, work, args, kwargs)
+
+ def _get(self):
+ tag, work, args, kwargs = None, None, None, None
+ while work is None:
+ self._delayfunc(self._wait_timeout)
+ tag = self._queue.get(self._timefunc())
+ item = self._work.get(tag, None)
+ if item is not None:
+ work, args, kwargs, interval = item
+ return (tag, work, args, kwargs, interval)
+
+ def __len__(self):
+ return len(self._work)
--
To view, visit http://gerrit.ovirt.org/29191
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ic06da1ba57868dc2c7db67a1868ad10087a1cff2
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Francesco Romani <fromani(a)redhat.com>
9 years, 4 months
Change in vdsm[master]: signals: Better handle signals to non-main threads
by dkuznets@redhat.com
Dima Kuznetsov has uploaded a new change for review.
Change subject: signals: Better handle signals to non-main threads
......................................................................
signals: Better handle signals to non-main threads
There is an issue with using signal.pause() and threads in python. The signal
might arrive to a non-main thread, and not get processed (because only the
main thread handles signals) until the main thread receives its own signal.
The issue is documented and tracked here:
https://bugzilla.redhat.com/show_bug.cgi?id=1114434
Change-Id: I5dbcd00cec22ef12f2b6253b016dcbd0aa889583
Signed-off-by: Dima Kuznetsov <dkuznets(a)redhat.com>
---
M lib/vdsm/utils.py
M vdsm/supervdsmServer
M vdsm/vdsm
3 files changed, 54 insertions(+), 2 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/92/29392/1
diff --git a/lib/vdsm/utils.py b/lib/vdsm/utils.py
index 2ba5762..35c8387 100644
--- a/lib/vdsm/utils.py
+++ b/lib/vdsm/utils.py
@@ -1141,3 +1141,55 @@
def prependDefer(self, func, *args, **kwargs):
self._finally.insert(0, (func, args, kwargs))
+
+
+_signalPoller = None
+
+
+def _getSignalPoller():
+ '''
+ This function creates a select.poll object that can be used in the same
+ manner as signal.pause(). The poll object returns each time a signal was
+ received by the process.
+
+ This function has to be called from the main thread.
+ '''
+ if _signalPoller is None:
+ (fd_r, fd_w) = os.pipe()
+
+ # The write end is going to be written to from a c-level signal
+ # handler, so it is a good idea to make sure it does not block.
+ flags = fcntl.fcntl(fd_w, fcntl.F_GETFL, 0)
+ flags |= os.O_NONBLOCK
+ fcntl.fcntl(fd_w, fcntl.F_SETFL, flags)
+
+ # This call must be done from the main thread.
+ signal.set_wakeup_fd(fd_w)
+
+ poller = select.poll()
+ poller.register(fd_r, select.POLLIN)
+
+ global _signalPoller
+ _signalPoller = poller
+
+ return _signalPoller
+
+
+def waitForSignal():
+ '''
+ This function acts like signal.pause(), it returns after a signal was
+ received and handled. Unlike signal.pause(), it will wake up even if other
+ thread caught the signal while this function was called.
+ '''
+ poller = _getSignalPoller()
+ try:
+ events = poller.poll()
+ for (fd, flags) in events:
+ if flags & select.POLLIN:
+ os.read(fd, 1)
+ except select.error as e:
+ # Signal was received in this thread while poll() was active
+ if (e.args[0] == errno.EINTR):
+ return
+ logging.error('Received error while waiting for signal',
+ exc_info=True)
diff --git a/vdsm/supervdsmServer b/vdsm/supervdsmServer
index a63caa2..0b0a13f 100755
--- a/vdsm/supervdsmServer
+++ b/vdsm/supervdsmServer
@@ -446,7 +446,7 @@
sourceroutethread.start()
while _running:
- signal.pause()
+ utils.waitForSignal()
log.debug("Terminated normally")
finally:
diff --git a/vdsm/vdsm b/vdsm/vdsm
index cbc16ff..9752320 100755
--- a/vdsm/vdsm
+++ b/vdsm/vdsm
@@ -83,7 +83,7 @@
cif.start()
try:
while running[0]:
- signal.pause()
+ utils.waitForSignal()
profile.stop()
finally:
--
To view, visit http://gerrit.ovirt.org/29392
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I5dbcd00cec22ef12f2b6253b016dcbd0aa889583
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Dima Kuznetsov <dkuznets(a)redhat.com>
9 years, 4 months
Change in vdsm[master]: Move multipath configuration to vdsm-tool configurator
by ykaplan@redhat.com
Yeela Kaplan has uploaded a new change for review.
Change subject: Move multipath configuration to vdsm-tool configurator
......................................................................
Move multipath configuration to vdsm-tool configurator
Previously multipath is reconfigured on each vdsm
service restart.
Now it will be reconfigured only on user demand.
Change-Id: Ife045908dc6e2aea9829b51482b909af1faf79da
Signed-off-by: Yeela Kaplan <ykaplan(a)redhat.com>
---
M lib/vdsm/tool/configurator.py
M vdsm/storage/hsm.py
M vdsm/storage/multipath.py
M vdsm/supervdsmServer
4 files changed, 147 insertions(+), 121 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/09/30909/1
diff --git a/lib/vdsm/tool/configurator.py b/lib/vdsm/tool/configurator.py
index 1e446a6..57923e8 100644
--- a/lib/vdsm/tool/configurator.py
+++ b/lib/vdsm/tool/configurator.py
@@ -26,6 +26,7 @@
import rpm
import shutil
import sys
+import tempfile
import traceback
import uuid
@@ -665,9 +666,155 @@
return configured
+class MultipathModuleConfigure(_ModuleConfigure):
+
+ _MPATH_CONF = "/etc/multipath.conf"
+
+ _STRG_MPATH_CONF = (
+ "\n\n"
+ "defaults {\n"
+ " polling_interval 5\n"
+ " getuid_callout \"%(scsi_id_path)s --whitelisted "
+ "--replace-whitespace --device=/dev/%%n\"\n"
+ " no_path_retry fail\n"
+ " user_friendly_names no\n"
+ " flush_on_last_del yes\n"
+ " fast_io_fail_tmo 5\n"
+ " dev_loss_tmo 30\n"
+ " max_fds 4096\n"
+ "}\n"
+ "\n"
+ "devices {\n"
+ "device {\n"
+ " vendor \"HITACHI\"\n"
+ " product \"DF.*\"\n"
+ " getuid_callout \"%(scsi_id_path)s --whitelisted "
+ "--replace-whitespace --device=/dev/%%n\"\n"
+ "}\n"
+ "device {\n"
+ " vendor \"COMPELNT\"\n"
+ " product \"Compellent Vol\"\n"
+ " no_path_retry fail\n"
+ "}\n"
+ "}"
+ )
+
+ _MAX_CONF_COPIES = 5
+
+ _OLD_TAGS = ["# RHAT REVISION 0.2", "# RHEV REVISION 0.3",
+ "# RHEV REVISION 0.4", "# RHEV REVISION 0.5",
+ "# RHEV REVISION 0.6", "# RHEV REVISION 0.7",
+ "# RHEV REVISION 0.8", "# RHEV REVISION 0.9"]
+
+ _MPATH_CONF_TAG = "# RHEV REVISION 1.0"
+ _MPATH_CONF_PRIVATE_TAG = "# RHEV PRIVATE"
+
+ _MPATH_CONF_TEMPLATE = _MPATH_CONF_TAG + _STRG_MPATH_CONF
+
+ _scsi_id = utils.CommandPath("scsi_id",
+ "/sbin/scsi_id", # EL6
+ "/usr/lib/udev/scsi_id", # Fedora
+ "/lib/udev/scsi_id", # Ubuntu
+ )
+
+ def getName(self):
+ return 'multipath'
+
+ def getServices(self):
+ return ["multipathd"]
+
+ def configure(self):
+ """
+ Set up the multipath daemon configuration to the known and
+ supported state. The original configuration, if any, is saved
+ """
+ if os.getuid() != 0:
+ raise UserWarning("Must run as root")
+
+ if self.isconfigured():
+ return
+
+ if os.path.exists(self._MPATH_CONF):
+ utils.rotateFiles(
+ os.path.dirname(self._MPATH_CONF),
+ os.path.basename(self._MPATH_CONF),
+ self._MAX_CONF_COPIES,
+ cp=True, persist=True)
+ with tempfile.NamedTemporaryFile() as f:
+ f.write(self._MPATH_CONF_TEMPLATE %
+ {'scsi_id_path': self._scsi_id.cmd})
+ f.flush()
+ cmd = [constants.EXT_CP, f.name,
+ self._MPATH_CONF]
+ rc, out, err = utils.execCmd(cmd)
+
+ if rc != 0:
+ raise RuntimeError("Failed to perform Multipath config.")
+ utils.persistFile(self._MPATH_CONF)
+
+ # Flush all unused multipath device maps
+ utils.execCmd([constants.EXT_MULTIPATH, "-F"])
+
+ cmd = [constants.EXT_VDSM_TOOL, "service-reload", "multipathd"]
+ rc, out, err = utils.execCmd(cmd)
+ if rc != 0:
+ sys.stdout.write("Failed to reload Multipath.")
+
+ def isconfigured(self, *args):
+ """
+ Check the multipath daemon configuration. The configuration file
+ /etc/multipath.conf should contain private tag in form
+ "RHEV REVISION X.Y" for this check to succeed.
+ If the tag above is followed by tag "RHEV PRIVATE" the configuration
+ should be preserved at all cost.
+ """
+ if os.getuid() != 0:
+ raise UserWarning("Must run as root")
+
+ if os.path.exists(self._MPATH_CONF):
+ first = second = ''
+ with open(self._MPATH_CONF) as f:
+ mpathconf = [x.strip("\n") for x in f.readlines()]
+ try:
+ first = mpathconf[0]
+ second = mpathconf[1]
+ except IndexError:
+ pass
+ if self._MPATH_CONF_PRIVATE_TAG in second:
+ sys.stdout.write("Manual override for multipath.conf detected"
+ " - preserving current configuration")
+ if self._MPATH_CONF_TAG not in first:
+ sys.stdout.write("This manual override for multipath.conf "
+ "was based on downrevved template. "
+ "You are strongly advised to "
+ "contact your support representatives")
+ return True
+
+ if self._MPATH_CONF_TAG in first:
+ sys.stdout.write("Current revision of multipath.conf detected,"
+ " preserving")
+ return True
+
+ for tag in self._OLD_TAGS:
+ if tag in first:
+ sys.stdout.write("Downrev multipath.conf detected, "
+ "upgrade required")
+ return False
+
+ sys.stdout.write("multipath Defaulting to False")
+ return False
+
+ def validate(self):
+ return self.isconfigured()
+
+ def reconfigureOnForce(self):
+ return self.isconfigured()
+
+
__configurers = (
LibvirtModuleConfigure(),
SanlockModuleConfigure(),
+ MultipathModuleConfigure(),
)
diff --git a/vdsm/storage/hsm.py b/vdsm/storage/hsm.py
index 320a50a..49f3680 100644
--- a/vdsm/storage/hsm.py
+++ b/vdsm/storage/hsm.py
@@ -355,9 +355,6 @@
self._preparedVolumes = defaultdict(list)
- if not multipath.isEnabled():
- multipath.setupMultipath()
-
self.__validateLvmLockingType()
oop.setDefaultImpl(config.get('irs', 'oop_impl'))
diff --git a/vdsm/storage/multipath.py b/vdsm/storage/multipath.py
index d626305..0c05409 100644
--- a/vdsm/storage/multipath.py
+++ b/vdsm/storage/multipath.py
@@ -25,7 +25,6 @@
import os
import errno
from glob import glob
-import tempfile
import logging
import re
from collections import namedtuple
@@ -37,53 +36,11 @@
import supervdsm
import devicemapper
-import storage_exception as se
-
DEV_ISCSI = "iSCSI"
DEV_FCP = "FCP"
DEV_MIXED = "MIXED"
-MAX_CONF_COPIES = 5
-
TOXIC_CHARS = '()*+?|^$.\\'
-
-MPATH_CONF = "/etc/multipath.conf"
-
-OLD_TAGS = ["# RHAT REVISION 0.2", "# RHEV REVISION 0.3",
- "# RHEV REVISION 0.4", "# RHEV REVISION 0.5",
- "# RHEV REVISION 0.6", "# RHEV REVISION 0.7",
- "# RHEV REVISION 0.8", "# RHEV REVISION 0.9"]
-MPATH_CONF_TAG = "# RHEV REVISION 1.0"
-MPATH_CONF_PRIVATE_TAG = "# RHEV PRIVATE"
-STRG_MPATH_CONF = (
- "\n\n"
- "defaults {\n"
- " polling_interval 5\n"
- " getuid_callout \"%(scsi_id_path)s --whitelisted "
- "--replace-whitespace --device=/dev/%%n\"\n"
- " no_path_retry fail\n"
- " user_friendly_names no\n"
- " flush_on_last_del yes\n"
- " fast_io_fail_tmo 5\n"
- " dev_loss_tmo 30\n"
- " max_fds 4096\n"
- "}\n"
- "\n"
- "devices {\n"
- "device {\n"
- " vendor \"HITACHI\"\n"
- " product \"DF.*\"\n"
- " getuid_callout \"%(scsi_id_path)s --whitelisted "
- "--replace-whitespace --device=/dev/%%n\"\n"
- "}\n"
- "device {\n"
- " vendor \"COMPELNT\"\n"
- " product \"Compellent Vol\"\n"
- " no_path_retry fail\n"
- "}\n"
- "}"
-)
-MPATH_CONF_TEMPLATE = MPATH_CONF_TAG + STRG_MPATH_CONF
log = logging.getLogger("Storage.Multipath")
@@ -108,76 +65,6 @@
# Now let multipath daemon pick up new devices
misc.execCmd([constants.EXT_MULTIPATH], sudo=True)
-
-
-def isEnabled():
- """
- Check the multipath daemon configuration. The configuration file
- /etc/multipath.conf should contain private tag in form
- "RHEV REVISION X.Y" for this check to succeed.
- If the tag above is followed by tag "RHEV PRIVATE" the configuration
- should be preserved at all cost.
-
- """
- if os.path.exists(MPATH_CONF):
- first = second = ''
- svdsm = supervdsm.getProxy()
- mpathconf = svdsm.readMultipathConf()
- try:
- first = mpathconf[0]
- second = mpathconf[1]
- except IndexError:
- pass
- if MPATH_CONF_PRIVATE_TAG in second:
- log.info("Manual override for multipath.conf detected - "
- "preserving current configuration")
- if MPATH_CONF_TAG not in first:
- log.warning("This manual override for multipath.conf "
- "was based on downrevved template. "
- "You are strongly advised to "
- "contact your support representatives")
- return True
-
- if MPATH_CONF_TAG in first:
- log.debug("Current revision of multipath.conf detected, "
- "preserving")
- return True
-
- for tag in OLD_TAGS:
- if tag in first:
- log.info("Downrev multipath.conf detected, upgrade required")
- return False
-
- log.debug("multipath Defaulting to False")
- return False
-
-
-def setupMultipath():
- """
- Set up the multipath daemon configuration to the known and
- supported state. The original configuration, if any, is saved
- """
- if os.path.exists(MPATH_CONF):
- utils.rotateFiles(
- os.path.dirname(MPATH_CONF),
- os.path.basename(MPATH_CONF), MAX_CONF_COPIES,
- cp=True, persist=True)
- with tempfile.NamedTemporaryFile() as f:
- f.write(MPATH_CONF_TEMPLATE % {'scsi_id_path': _scsi_id.cmd})
- f.flush()
- cmd = [constants.EXT_CP, f.name, MPATH_CONF]
- rc = misc.execCmd(cmd, sudo=True)[0]
- if rc != 0:
- raise se.MultipathSetupError()
- utils.persistFile(MPATH_CONF)
-
- # Flush all unused multipath device maps
- misc.execCmd([constants.EXT_MULTIPATH, "-F"], sudo=True)
-
- cmd = [constants.EXT_VDSM_TOOL, "service-reload", "multipathd"]
- rc = misc.execCmd(cmd, sudo=True)[0]
- if rc != 0:
- raise se.MultipathReloadError()
def deduceType(a, b):
diff --git a/vdsm/supervdsmServer b/vdsm/supervdsmServer
index aa58a13..4ab65f6 100755
--- a/vdsm/supervdsmServer
+++ b/vdsm/supervdsmServer
@@ -139,11 +139,6 @@
return _getDevicePartedInfo(*args, **kwargs)
@logDecorator
- def readMultipathConf(self):
- with open(MPATH_CONF) as f:
- return [x.strip("\n") for x in f.readlines()]
-
- @logDecorator
def getScsiSerial(self, *args, **kwargs):
return _getScsiSerial(*args, **kwargs)
--
To view, visit http://gerrit.ovirt.org/30909
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ife045908dc6e2aea9829b51482b909af1faf79da
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Yeela Kaplan <ykaplan(a)redhat.com>
9 years, 5 months
Change in vdsm[master]: configurator: move usermod to spec
by Douglas Schilling Landgraf
Douglas Schilling Landgraf has uploaded a new change for review.
Change subject: configurator: move usermod to spec
......................................................................
configurator: move usermod to spec
vdsm-tool configure --force adds to /etc/group (qemu/kvm) sanlock
but doesn't persist the file in ovirt node distro which will
affect vdsm start on next reboot. This patch moves the usermod to
spec file.
Change-Id: I668552fa037414e9a6aee5b049d61749268f85d0
Signed-off-by: Douglas Schilling Landgraf <dougsland(a)redhat.com>
---
M lib/vdsm/tool/configurator.py
M vdsm.spec.in
2 files changed, 5 insertions(+), 25 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/55/26055/1
diff --git a/lib/vdsm/tool/configurator.py b/lib/vdsm/tool/configurator.py
index d1c876c..516a7bc 100644
--- a/lib/vdsm/tool/configurator.py
+++ b/lib/vdsm/tool/configurator.py
@@ -24,7 +24,7 @@
from .. import utils
from . import service, expose
-from ..constants import P_VDSM_EXEC, QEMU_PROCESS_GROUP, VDSM_GROUP
+from ..constants import P_VDSM_EXEC
class _ModuleConfigure(object):
@@ -108,8 +108,6 @@
class SanlockModuleConfigure(_ModuleConfigure):
- SANLOCK_GROUPS = (QEMU_PROCESS_GROUP, VDSM_GROUP)
-
def __init__(self):
super(SanlockModuleConfigure, self).__init__()
@@ -118,28 +116,6 @@
def getServices(self):
return ['sanlock']
-
- def configure(self):
- """
- Configure sanlock process groups
- """
- if os.getuid() != 0:
- raise UserWarning("Must run as root")
-
- rc, out, err = utils.execCmd(
- (
- '/usr/sbin/usermod',
- '-a',
- '-G',
- ','.join(self.SANLOCK_GROUPS),
- 'sanlock'
- ),
- raw=True,
- )
- sys.stdout.write(out)
- sys.stderr.write(err)
- if rc != 0:
- raise RuntimeError("Failed to perform sanlock config.")
def isconfigured(self):
"""
diff --git a/vdsm.spec.in b/vdsm.spec.in
index 361a9c1..b149568 100644
--- a/vdsm.spec.in
+++ b/vdsm.spec.in
@@ -690,6 +690,7 @@
/usr/sbin/useradd -r -u 36 -g %{vdsm_group} -d /var/lib/vdsm \
-s /sbin/nologin -c "Node Virtualization Manager" %{vdsm_user}
/usr/sbin/usermod -a -G %{qemu_group},%{snlk_group} %{vdsm_user}
+/usr/sbin/usermod -a -G %{qemu_group},%{vdsm_group} %{snlk_user}
%post
%{_bindir}/vdsm-tool sebool-config || :
@@ -734,6 +735,9 @@
%endif
%preun
+/usr/bin/gpasswd -d %{snlk_user} %{qemu_group}
+/usr/bin/gpasswd -d %{snlk_user} %{vdsm_group}
+
if [ "$1" -eq 0 ]; then
start_conf_section="## beginning of configuration section by vdsm"
end_conf_section="## end of configuration section by vdsm"
--
To view, visit http://gerrit.ovirt.org/26055
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I668552fa037414e9a6aee5b049d61749268f85d0
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Douglas Schilling Landgraf <dougsland(a)redhat.com>
9 years, 6 months
Change in vdsm[master]: vdsm-tool: add the cmd get-engine-web-ca-issuer
by Douglas Schilling Landgraf
Douglas Schilling Landgraf has uploaded a new change for review.
Change subject: vdsm-tool: add the cmd get-engine-web-ca-issuer
......................................................................
vdsm-tool: add the cmd get-engine-web-ca-issuer
If oVirt Node is registered via oVirt Engine currently
we don't show to users which engine the node is registered.
This patch extracts the issuer field from engine web ca to use
the CN (commonname) data in ovirt-node-plugin-vdsm.
Change-Id: I70a19f54169498c53cabba0ac75e7a5611762ece
Bug-Url: https://bugzilla.redhat.com/show_bug.cgi?id=1039267
Signed-off-by: Douglas Schilling Landgraf <dougsland(a)redhat.com>
---
M lib/vdsm/tool/validate_ovirt_certs.py.in
1 file changed, 15 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/51/27651/1
diff --git a/lib/vdsm/tool/validate_ovirt_certs.py.in b/lib/vdsm/tool/validate_ovirt_certs.py.in
index 6a8c13f..149dc30 100644
--- a/lib/vdsm/tool/validate_ovirt_certs.py.in
+++ b/lib/vdsm/tool/validate_ovirt_certs.py.in
@@ -35,6 +35,7 @@
PATH_CERT = '/etc/pki/vdsm/certs/'
PATH_CACERT = PATH_CERT + 'cacert.pem'
PATH_VDSMCERT = PATH_CERT + 'vdsmcert.pem'
+PATH_ENGINE_WEB_CA = PATH_CERT + 'engine_web_ca.pem'
def is_our_cafile(cacert):
@@ -60,3 +61,17 @@
os.chown(PATH_CACERT, uid, gid)
ovirtfunctions.ovirt_store_config(PATH_CACERT)
break
+
+
+@expose('get-engine-web-ca-issuer')
+def get_engine_web_ca_issuer():
+ """
+ Extract from engine web ca certificate the issuer field
+ """
+ if os.path.exists(PATH_ENGINE_WEB_CA):
+ cmd = [
+ EX_OPENSSL, 'x509', '-in', PATH_ENGINE_WEB_CA, '-noout', '-issuer'
+ ]
+ rc, out, err = execCmd(cmd, raw=True)
+ if not err:
+ print(out),
--
To view, visit http://gerrit.ovirt.org/27651
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I70a19f54169498c53cabba0ac75e7a5611762ece
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Douglas Schilling Landgraf <dougsland(a)redhat.com>
9 years, 6 months