vdsm-patches June 2014

vdsm-patches@lists.fedorahosted.org

41 participants
740 discussions

Change in vdsm[master]: This commit adds a man page for the vdsm-tool command line u...
by Jenkins CI RO 05 Jun '14

05 Jun '14

oVirt Jenkins CI Server has posted comments on this change. Change subject: This commit adds a man page for the vdsm-tool command line utility. ...................................................................... Patch Set 3: Build Successful http://jenkins.ovirt.org/job/vdsm_master_install_rpm_sanity_gerrit/687/ : SUCCESS http://jenkins.ovirt.org/job/vdsm_master_create-rpms_merged/1391/ : SUCCESS -- To view, visit http://gerrit.ovirt.org/28196 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: comment Gerrit-Change-Id: Ieb3e4ab7947a151a30ccc1c1a41e8df9d02d1d65 Gerrit-PatchSet: 3 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Andrew Dahms <andrewjdahms(a)gmail.com> Gerrit-Reviewer: Antoni Segura Puimedon <asegurap(a)redhat.com> Gerrit-Reviewer: Dan Kenigsberg <danken(a)redhat.com> Gerrit-Reviewer: Yaniv Bronhaim <ybronhei(a)redhat.com> Gerrit-Reviewer: automation(a)ovirt.org Gerrit-Reviewer: mooli tayer <mtayer(a)redhat.com> Gerrit-Reviewer: oVirt Jenkins CI Server Gerrit-HasComments: No

1 0

Change in vdsm[master]: json-rpc: Protocol detection
by Nir Soffer 05 Jun '14

05 Jun '14

Nir Soffer has posted comments on this change. Change subject: json-rpc: Protocol detection ...................................................................... Patch Set 33: (6 comments) protocol detection and xmlrpc looks fine, expect two issues: - This patch exposes a bug we had in the non-secure xmlprc version, where http 1.1 is not used. This causes a regression in the secure xmlrpc version. Trivial fix, but requires some testing with ssl=false. - None return value from SSLSocket.recv is not handled - trivial fix. http://gerrit.ovirt.org/#/c/26300/33/vdsm/BindingXMLRPC.py File vdsm/BindingXMLRPC.py: Line 90: HTTP_HEADER_FLOWID = "FlowID" Line 91: Line 92: threadLocal = self.cif.threadLocal Line 93: Line 94: basehandler = SimpleXMLRPCServer.SimpleXMLRPCRequestHandler This change expose a bug in the non-secure xmlrpc version. SecureXMLRPCServer.SecureXMLRPCRequestHandler used before this patch is actually utils.IPXMLRPCRequestHandler, which is modified to support HTTP/1.1 since ovirt-3.4. So the correct code here would to remove the basehandler variable and replace its usage with utils.IPXMLRPCRequestHandler. This requires testing vdsm with ssl=false in vdsm.conf. Line 95: Line 96: class RequestHandler(basehandler): Line 97: Line 98: # Timeout for the request socket Line 92: threadLocal = self.cif.threadLocal Line 93: Line 94: basehandler = SimpleXMLRPCServer.SimpleXMLRPCRequestHandler Line 95: Line 96: class RequestHandler(basehandler): basehandler -> utils.IPXMLRPCRequestHandler Line 97: Line 98: # Timeout for the request socket Line 99: timeout = 60 Line 100: log = logging.getLogger("BindingXMLRPC.RequestHandler") Line 115: Line 116: def setup(self): Line 117: threadLocal.client = self.client_address[0] Line 118: threadLocal.server = self.request.getsockname()[0] Line 119: return basehandler.setup(self) basehandler -> utils.IPXMLRPCRequestHandler Line 120: Line 121: def do_GET(self): Line 122: try: Line 123: length = self._getIntHeader(self.HEADER_SIZE, Line 258: self.end_headers() Line 259: self.wfile.write(json_response) Line 260: Line 261: def parse_request(self): Line 262: r = basehandler.parse_request(self) basehandler -> utils.IPXMLRPCRequestHandler Line 263: threadLocal.flowID = self.headers.get(HTTP_HEADER_FLOWID) Line 264: return r Line 265: Line 266: def finish(self): Line 263: threadLocal.flowID = self.headers.get(HTTP_HEADER_FLOWID) Line 264: return r Line 265: Line 266: def finish(self): Line 267: basehandler.finish(self) basehandler -> utils.IPXMLRPCRequestHandler Line 268: threadLocal.client = None Line 269: threadLocal.server = None Line 270: threadLocal.flowID = None Line 271: http://gerrit.ovirt.org/#/c/26300/33/vdsm/protocolDetector.py File vdsm/protocolDetector.py: Line 178: self.log.warning("Unable to read data") Line 179: self._remove_connection(client_socket) Line 180: client_socket.close() Line 181: return Line 182: Looking in the betterAsyncore.py, we should handle here the case where data is None. This happens when using M2Crypto, if the SSLSocket want to read more data from the other side before it can return data. I would add this for now: if data is None: # SSLSocket wants to read more data from the peer return Line 183: self._remove_connection(client_socket) Line 184: try: Line 185: handler = self.detect_protocol(data) Line 186: except _CannotDetectProtocol: -- To view, visit http://gerrit.ovirt.org/26300 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: comment Gerrit-Change-Id: Id739a40e2b37dcc175137ec91cd5ec166ad24a75 Gerrit-PatchSet: 33 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Piotr Kliczewski <piotr.kliczewski(a)gmail.com> Gerrit-Reviewer: Barak Azulay <bazulay(a)redhat.com> Gerrit-Reviewer: Francesco Romani <fromani(a)redhat.com> Gerrit-Reviewer: Liron Ar <laravot(a)redhat.com> Gerrit-Reviewer: Michal Skrivanek <michal.skrivanek(a)redhat.com> Gerrit-Reviewer: Nir Soffer <nsoffer(a)redhat.com> Gerrit-Reviewer: Piotr Kliczewski <piotr.kliczewski(a)gmail.com> Gerrit-Reviewer: Saggi Mizrahi <smizrahi(a)redhat.com> Gerrit-Reviewer: Vinzenz Feenstra <vfeenstr(a)redhat.com> Gerrit-Reviewer: Yaniv Bronhaim <ybronhei(a)redhat.com> Gerrit-Reviewer: Yeela Kaplan <ykaplan(a)redhat.com> Gerrit-Reviewer: automation(a)ovirt.org Gerrit-Reviewer: oVirt Jenkins CI Server Gerrit-HasComments: Yes

1 0

Change in vdsm[master]: This commit adds a man page for the vdsm-tool command line u...
by Dan Kenigsberg 05 Jun '14

05 Jun '14

Dan Kenigsberg has submitted this change and it was merged. Change subject: This commit adds a man page for the vdsm-tool command line utility. ...................................................................... This commit adds a man page for the vdsm-tool command line utility. This commit adds the file vdsm-tool.1.in to the vdsm-tool directory, updates the vdsm-tool make file and the vdsm spec file to include and install the new man page for vdsm-tool to the /usr/share/man/man1 directory. Change-Id: Ieb3e4ab7947a151a30ccc1c1a41e8df9d02d1d65 Signed-off-by: Andrew Dahms <andrewjdahms(a)gmail.com> Reviewed-on: http://gerrit.ovirt.org/28196 Reviewed-by: Dan Kenigsberg <danken(a)redhat.com> Tested-by: Dan Kenigsberg <danken(a)redhat.com> --- M vdsm-tool/Makefile.am A vdsm-tool/vdsm-tool.1.in M vdsm.spec.in 3 files changed, 198 insertions(+), 0 deletions(-) Approvals: Dan Kenigsberg: Verified; Looks good to me, approved -- To view, visit http://gerrit.ovirt.org/28196 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: merged Gerrit-Change-Id: Ieb3e4ab7947a151a30ccc1c1a41e8df9d02d1d65 Gerrit-PatchSet: 3 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Andrew Dahms <andrewjdahms(a)gmail.com> Gerrit-Reviewer: Antoni Segura Puimedon <asegurap(a)redhat.com> Gerrit-Reviewer: Dan Kenigsberg <danken(a)redhat.com> Gerrit-Reviewer: Yaniv Bronhaim <ybronhei(a)redhat.com> Gerrit-Reviewer: automation(a)ovirt.org Gerrit-Reviewer: mooli tayer <mtayer(a)redhat.com> Gerrit-Reviewer: oVirt Jenkins CI Server

1 0

Change in vdsm[master]: This commit adds a man page for the vdsm-tool command line u...
by Dan Kenigsberg 05 Jun '14

05 Jun '14

Dan Kenigsberg has posted comments on this change. Change subject: This commit adds a man page for the vdsm-tool command line utility. ...................................................................... Patch Set 2: Verified+1 Code-Review+2 Copying score, only a minor text change. -- To view, visit http://gerrit.ovirt.org/28196 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: comment Gerrit-Change-Id: Ieb3e4ab7947a151a30ccc1c1a41e8df9d02d1d65 Gerrit-PatchSet: 2 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Andrew Dahms <andrewjdahms(a)gmail.com> Gerrit-Reviewer: Antoni Segura Puimedon <asegurap(a)redhat.com> Gerrit-Reviewer: Dan Kenigsberg <danken(a)redhat.com> Gerrit-Reviewer: Yaniv Bronhaim <ybronhei(a)redhat.com> Gerrit-Reviewer: automation(a)ovirt.org Gerrit-Reviewer: mooli tayer <mtayer(a)redhat.com> Gerrit-Reviewer: oVirt Jenkins CI Server Gerrit-HasComments: No

1 0

Change in vdsm[master]: misc: Faster version of readspeed
by Nir Soffer 05 Jun '14

05 Jun '14

Nir Soffer has uploaded a new change for review. Change subject: misc: Faster version of readspeed ...................................................................... misc: Faster version of readspeed After optimizing list2cmdline, we can see that the profile is dominated by utils.execCmd(), used to invoke dd for checking reading speed. This code is quite complex, using fancy classes such as AsyncProc and StreamWrapper. These are useful when you read from multiple streams to avoid deadlocks. However for checking reading speed, this seems like an overkill; we want dd to read a block from storage and read 3 lines of statistics from stderr. We don't care about the data read by dd, and have no reason to read this data from dd stdout. This patch uses cpopen.Popen directly, running dd reading from a path and writing to /dev/null, and use Popen.communicate to read dd stderr. Since we are communicating with dd using a single pipe, a deadlock is not possible. I tested this patch on two identical machines connected to 32 storage domains. Each one run 5 minutes as spm, and 5 minutes as hsm during 10 minutes profiling session. The first system was running the previous patch in this branch: Sun May 11 02:23:31 2014 vdsmd-list2cmdline.prof 8758092 function calls (8816678 primitive calls) in 81.525 CPU seconds Ordered by: internal time List reduced from 1872 to 20 due to restriction <20> ncalls tottime percall cumtime percall filename:lineno(function) 101/136 6.554 0.065 46.116 0.339 threading.py:481(Thread.run) 1352629 3.652 0.000 6.595 0.000 utils.py:557(AsyncProc._processStreams) 134289 2.296 0.000 2.296 0.000 pthread.py:98(Lock.unlock) 9764 2.282 0.000 3.617 0.000 __init__.py:226(LogRecord.__init__) 25631 2.202 0.000 10.448 0.000 utils.py:466(_streamWrapper._readNonBlock) 116992 2.145 0.000 2.157 0.000 pthread.py:95(Lock.lock) 6049 2.031 0.000 6.722 0.001 misc.py:125(findCaller) 18524 1.954 0.000 22.039 0.001 io.py:492(BufferedReader.readline) 58408 1.788 0.000 1.882 0.000 genericpath.py:85(_splitext) 2521 1.631 0.001 8.974 0.004 utils.py:623(AsyncProc.wait) Ordered by: cumulative time List reduced from 1872 to 20 due to restriction <20> ncalls tottime percall cumtime percall filename:lineno(function) 33/36 0.001 0.000 69.017 1.917 utils.py:770(wrapper) 31/32 0.072 0.002 68.740 2.148 domainMonitor.py:163(DomainMonitorThread._monitorLoop) 1072/2105 0.720 0.001 68.122 0.032 domainMonitor.py:186(DomainMonitorThread._monitorDomain) 2519 0.494 0.000 56.573 0.022 utils.py:666(execCmd) 2041/2105 0.110 0.000 46.814 0.022 blockSD.py:594(BlockStorageDomain.getReadDelay) 2105 0.139 0.000 46.411 0.022 misc.py:222(readspeed) 2105 0.314 0.000 46.241 0.022 misc.py:192(_readfile) 101/136 6.554 0.065 46.116 0.339 threading.py:481(Thread.run) 2516 0.133 0.000 34.884 0.014 utils.py:633(AsyncProc.communicate) 18520/18524 0.255 0.000 22.294 0.001 io.py:532(BufferedReader.next) The seconds system was running this branch: Sun May 11 02:23:38 2014 vdsmd-optimize.prof 2113774 function calls (2168006 primitive calls) in 43.489 CPU seconds Ordered by: internal time List reduced from 1873 to 20 due to restriction <20> ncalls tottime percall cumtime percall filename:lineno(function) 111/136 8.270 0.075 29.113 0.214 threading.py:481(Thread.run) 41106 0.912 0.000 0.912 0.000 pthread.py:95(Lock.lock) 2146 0.898 0.000 1.412 0.001 subprocess.py:1344(Popen._communicate_with_poll) 27318 0.866 0.000 1.873 0.000 utils.py:557(AsyncProc._processStreams) 1867 0.810 0.000 2.509 0.001 misc.py:126(findCaller) 43735 0.791 0.000 0.791 0.000 pthread.py:98(Lock.unlock) 8112 0.719 0.000 5.590 0.001 __init__.py:1204(Logger.callHandlers) 7326/7338 0.692 0.000 0.741 0.000 pthread.py:133(Cond.timedwait) 1553/2145 0.648 0.000 34.541 0.016 domainMonitor.py:186(DomainMonitorThread._monitorDomain) Ordered by: cumulative time List reduced from 1873 to 20 due to restriction <20> ncalls tottime percall cumtime percall filename:lineno(function) 34/36 0.001 0.000 35.231 0.979 utils.py:770(wrapper) 32 0.075 0.002 35.099 1.097 domainMonitor.py:163(DomainMonitorThread._monitorLoop) 1553/2145 0.648 0.000 34.541 0.016 domainMonitor.py:186(DomainMonitorThread._monitorDomain) 111/136 8.270 0.075 29.113 0.214 threading.py:481(Thread.run) 13501 0.179 0.000 17.309 0.001 sdc.py:51(DomainProxy.getRealDomain) 13501 0.381 0.000 17.129 0.001 sdc.py:101(StorageDomainCache._realProduce) 97 0.009 0.000 15.431 0.159 sdc.py:133(StorageDomainCache._findDomain) 97 0.007 0.000 15.112 0.156 blockSD.py:1359(findDomain) 91 0.007 0.000 14.304 0.157 sdc.py:148(StorageDomainCache._findUnfetchedDomain) 97 0.042 0.000 14.060 0.145 blockSD.py:400(BlockStorageDomain.__init__) ... 2145 0.344 0.000 8.410 0.004 misc.py:223(readspeed) Comparing the profiles show: - Number of function calls dropped from 8758092 to 2113774 (75%) - Number of cpu seconds dropped from 81 to 43 (46%) - Time spent in _monitorDomain() dropped from 68 to 34 seconds (%50) - Time spent in readspeed() dropped from 46.4 to 8.4 seconds (81%) Change-Id: I7ecf1f27d8434aeae672e92ec7adb12e52e419a9 Signed-off-by: Nir Soffer <nsoffer(a)redhat.com> --- M vdsm/storage/misc.py 1 file changed, 20 insertions(+), 7 deletions(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/53/27553/1 diff --git a/vdsm/storage/misc.py b/vdsm/storage/misc.py index 12d4d97..8520c37 100644 --- a/vdsm/storage/misc.py +++ b/vdsm/storage/misc.py @@ -32,6 +32,7 @@ from functools import wraps, partial from itertools import chain, imap import contextlib +import cpopen import errno import glob import logging @@ -219,26 +220,38 @@ ) -def readspeed(path, buffersize=None): +def readspeed(path, buffersize): """ Measures the amount of bytes transferred and the time elapsed reading the content of the file/device """ - rc, _, err = _readfile(path, buffersize) - if rc != 0: - log.error("Unable to read file '%s'", path) - raise se.MiscFileReadException(path) + # Read data from file to /dev/null, as we are not interested in the data + # itself, only in the read statistics provided by dd on stderr. + cmd = [constants.EXT_DD, "if=%s" % path, "iflag=%s" % DIRECTFLAG, + "of=/dev/null", "bs=%d" % buffersize, "count=1"] + + log.debug("Running %s", utils.list2cmdline(cmd)) + p = cpopen.CPopen(cmd, close_fds=True) + try: + out, err = p.communicate() + if p.returncode: + log.error("Unable to read file '%s'", path) + raise se.MiscFileReadException(path) + finally: + if p.returncode is None: + p.kill() try: # The statistics are located in the last line: - m = _readspeed_regex.match(err[-1]) + statsLine = err.splitlines()[-1] + m = _readspeed_regex.match(statsLine) except IndexError: log.error("Unable to find dd statistics to parse") raise se.MiscFileReadException(path) if m is None: - log.error("Unable to parse dd output: '%s'", err[-1]) + log.error("Unable to parse dd output: '%s'", statsLine) raise se.MiscFileReadException(path) return { -- To view, visit http://gerrit.ovirt.org/27553 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: I7ecf1f27d8434aeae672e92ec7adb12e52e419a9 Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com>

8 58

Change in vdsm[master]: This commit adds a man page for the vdsm-tool command line u...
by Dan Kenigsberg 05 Jun '14

05 Jun '14

Dan Kenigsberg has posted comments on this change. Change subject: This commit adds a man page for the vdsm-tool command line utility. ...................................................................... Patch Set 1: Code-Review-1 (1 comment) http://gerrit.ovirt.org/#/c/28196/1/vdsm-tool/vdsm-tool.1.in File vdsm-tool/vdsm-tool.1.in: Line 186: <Author list> Line 187: .SH BUGS Line 188: Report bugs to <http://bugzilla.redhat.com> Line 189: .SH COPYRIGHT Line 190: Copyright 2011 Red Hat, Inc. License GPLv2: GNU GPL Version 2 <http://gnu.org/licenses/gpl.html>. > shouldn't be 2014? that's the copyright of the code (not of the docs) so the years should be 2011-2014, I suppose. -- To view, visit http://gerrit.ovirt.org/28196 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: comment Gerrit-Change-Id: Ieb3e4ab7947a151a30ccc1c1a41e8df9d02d1d65 Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Andrew Dahms <andrewjdahms(a)gmail.com> Gerrit-Reviewer: Antoni Segura Puimedon <asegurap(a)redhat.com> Gerrit-Reviewer: Dan Kenigsberg <danken(a)redhat.com> Gerrit-Reviewer: Yaniv Bronhaim <ybronhei(a)redhat.com> Gerrit-Reviewer: automation(a)ovirt.org Gerrit-Reviewer: mooli tayer <mtayer(a)redhat.com> Gerrit-Reviewer: oVirt Jenkins CI Server Gerrit-HasComments: Yes

1 0

Change in vdsm[master]: api: expose prepare and teardown image
by Federico Simoncelli 05 Jun '14

05 Jun '14

Federico Simoncelli has uploaded a new change for review. Change subject: api: expose prepare and teardown image ...................................................................... api: expose prepare and teardown image Bug-Url: https://bugzilla.redhat.com/show_bug.cgi?id=1092166 Change-Id: I6a4e21513a2a888ff5ee4bc085a3da9a8bf8980b Signed-off-by: Federico Simoncelli <fsimonce(a)redhat.com> --- M client/vdsClient.py M vdsm/API.py M vdsm/BindingXMLRPC.py M vdsm_api/vdsmapi-schema.json 4 files changed, 75 insertions(+), 0 deletions(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/89/28389/1 diff --git a/client/vdsClient.py b/client/vdsClient.py index 0235837..80b8a0b 100644 --- a/client/vdsClient.py +++ b/client/vdsClient.py @@ -1340,6 +1340,18 @@ return 0, image['uuid'] + def prepareImage(self, args): + if len(args) < 3 or len(args) > 4: + raise ValueError('Wrong number of parameters') + ret = self.s.prepareImage(*args) + return ret['status']['code'], ret['status']['message'] + + def teardownImage(self, args): + if len(args) < 3 or len(args) > 4: + raise ValueError('Wrong number of parameters') + ret = self.s.teardownImage(*args) + return ret['status']['code'], ret['status']['message'] + def moveMultiImage(self, args): spUUID = args[0] srcDomUUID = args[1] @@ -2423,6 +2435,14 @@ 'Download an image from a remote endpoint using the specified', 'methodArgs.' )), + 'prepareImage': (serv.prepareImage, ( + '<spUUID> <sdUUID> <imgUUID> [<volUUID>]', + 'Prepare an image, making the needed volumes available.' + )), + 'teardownImage': (serv.teardownImage, ( + '<spUUID> <sdUUID> <imgUUID> [<volUUID>]', + 'Teardown an image, releasing the prepared volumes.' + )), 'moveMultiImage': (serv.moveMultiImage, ('<spUUID> <srcDomUUID> <dstDomUUID> ' '<imgList>({imgUUID=postzero,' diff --git a/vdsm/API.py b/vdsm/API.py index 3ed1a0c..8b1032d 100644 --- a/vdsm/API.py +++ b/vdsm/API.py @@ -848,6 +848,14 @@ return self._irs.downloadImage( methodArgs, self._spUUID, self._sdUUID, self._UUID, volUUID) + def prepare(self, volUUID=None): + return self._irs.prepareImage( + self._sdUUID, self._spUUID, self._UUID, volUUID) + + def teardown(self, volUUID=None): + return self._irs.teardownImage( + self._sdUUID, self._spUUID, self._UUID, volUUID) + def uploadToStream(self, methodArgs, callback, startEvent, volUUID=None): return self._irs.uploadImageToStream( methodArgs, callback, startEvent, self._spUUID, self._sdUUID, diff --git a/vdsm/BindingXMLRPC.py b/vdsm/BindingXMLRPC.py index 3cb4af7..87ce86f 100644 --- a/vdsm/BindingXMLRPC.py +++ b/vdsm/BindingXMLRPC.py @@ -686,6 +686,14 @@ image = API.Image(imgUUID, spUUID, sdUUID) return image.download(methodArgs, volUUID) + def imagePrepare(self, spUUID, sdUUID, imgUUID, volUUID=None): + image = API.Image(imgUUID, spUUID, sdUUID) + return image.prepare(volUUID) + + def imageTeardown(self, spUUID, sdUUID, imgUUID, volUUID=None): + image = API.Image(imgUUID, spUUID, sdUUID) + return image.teardown(volUUID) + def poolConnect(self, spUUID, hostID, scsiKey, msdUUID, masterVersion, domainsMap=None, options=None): pool = API.StoragePool(spUUID) @@ -1032,6 +1040,8 @@ (self.imageSyncData, 'syncImageData'), (self.imageUpload, 'uploadImage'), (self.imageDownload, 'downloadImage'), + (self.imagePrepare, 'prepareImage'), + (self.imageTeardown, 'teardownImage'), (self.poolConnect, 'connectStoragePool'), (self.poolConnectStorageServer, 'connectStorageServer'), (self.poolCreate, 'createStoragePool'), diff --git a/vdsm_api/vdsmapi-schema.json b/vdsm_api/vdsmapi-schema.json index 993c7f4..1d094b8 100644 --- a/vdsm_api/vdsmapi-schema.json +++ b/vdsm_api/vdsmapi-schema.json @@ -4162,6 +4162,43 @@ 'storagedomainID': 'UUID', 'imageID': 'UUID', '*volumeID': 'UUID'}, 'returns': 'UUID'} +## +# @Image.prepare: +# +# Prepare an image, making the needed volumes available. +# +# @storagepoolID: The UUID of the Storage Pool associated with the Image +# +# @storagedomainID: The UUID of the Storage Domain associated with the Image +# +# @imageID: The UUID of the Image +# +# @volumeID: #optional The UUID of the Volume (optional) +# +# Since: 4.15.0 +## +{'command': {'class': 'Image', 'name': 'prepare'}, + 'data': {'storagepoolID': 'UUID', 'storagedomainID': 'UUID', + 'imageID': 'UUID', '*volumeID': 'UUID'}} + +## +# @Image.teardown: +# +# Teardown an image, releasing the prepared volumes. +# +# @storagepoolID: The UUID of the Storage Pool associated with the Image +# +# @storagedomainID: The UUID of the Storage Domain associated with the Image +# +# @imageID: The UUID of the Image +# +# @volumeID: #optional The UUID of the Volume (optional) +# +# Since: 4.15.0 +## +{'command': {'class': 'Image', 'name': 'teardown'}, + 'data': {'storagepoolID': 'UUID', 'storagedomainID': 'UUID', + 'imageID': 'UUID', '*volumeID': 'UUID'}} ## Category: @LVMVolumeGroup ################################################### ## -- To view, visit http://gerrit.ovirt.org/28389 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: I6a4e21513a2a888ff5ee4bc085a3da9a8bf8980b Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Federico Simoncelli <fsimonce(a)redhat.com>

4 6

Change in vdsm[master]: json-rpc: Protocol detection
by ybronhei＠redhat.com 05 Jun '14

05 Jun '14

Yaniv Bronhaim has posted comments on this change. Change subject: json-rpc: Protocol detection ...................................................................... Patch Set 33: Code-Review+1 please move BindingXMLRPC.py to rpc dir in another patch (this already tested) -- To view, visit http://gerrit.ovirt.org/26300 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: comment Gerrit-Change-Id: Id739a40e2b37dcc175137ec91cd5ec166ad24a75 Gerrit-PatchSet: 33 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Piotr Kliczewski <piotr.kliczewski(a)gmail.com> Gerrit-Reviewer: Barak Azulay <bazulay(a)redhat.com> Gerrit-Reviewer: Francesco Romani <fromani(a)redhat.com> Gerrit-Reviewer: Liron Ar <laravot(a)redhat.com> Gerrit-Reviewer: Michal Skrivanek <michal.skrivanek(a)redhat.com> Gerrit-Reviewer: Nir Soffer <nsoffer(a)redhat.com> Gerrit-Reviewer: Piotr Kliczewski <piotr.kliczewski(a)gmail.com> Gerrit-Reviewer: Saggi Mizrahi <smizrahi(a)redhat.com> Gerrit-Reviewer: Vinzenz Feenstra <vfeenstr(a)redhat.com> Gerrit-Reviewer: Yaniv Bronhaim <ybronhei(a)redhat.com> Gerrit-Reviewer: Yeela Kaplan <ykaplan(a)redhat.com> Gerrit-Reviewer: automation(a)ovirt.org Gerrit-Reviewer: oVirt Jenkins CI Server Gerrit-HasComments: No

1 0

Change in vdsm[master]: jsonrpc: Stomp support
by smizrahi＠redhat.com 05 Jun '14

05 Jun '14

Saggi Mizrahi has uploaded a new change for review. Change subject: jsonrpc: Stomp support ...................................................................... jsonrpc: Stomp support The mini broker turned up to be much more work than I expected. This is an intermediate solution. VDSM will accept broker like message but will ignore most of them and not enforce policy. This is so that we can get the engine talking in the correct way and work on the mini broker stress free. This also means that we can release VDSM without the mini broker using a message format that is future ready. It is expected from the engine to send the CONNECT and SUBSCRIBE frames even though VDSM doesn't yet enforce them because future VDSM's probably will. Change-Id: I22bcae1e150dea7bc7d9fecefb6847c48bfe8949 Signed-off-by: Saggi Mizrahi <smizrahi(a)redhat.com> --- M lib/yajsonrpc/__init__.py M lib/yajsonrpc/betterAsyncore.py A lib/yajsonrpc/stomp.py A lib/yajsonrpc/stompReactor.py M tests/jsonRpcTests.py M tests/jsonRpcUtils.py M vdsm_api/BindingJsonRpc.py 7 files changed, 895 insertions(+), 1 deletion(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/50/26750/1 diff --git a/lib/yajsonrpc/__init__.py b/lib/yajsonrpc/__init__.py index fabc5d4..34ff97f 100644 --- a/lib/yajsonrpc/__init__.py +++ b/lib/yajsonrpc/__init__.py @@ -466,6 +466,7 @@ self._threadFactory = threadFactory def queueRequest(self, req): + self.log.debug("Queueing request") self._workQueue.put_nowait(req) def _serveRequest(self, ctx, req): @@ -502,11 +503,14 @@ def serve_requests(self): while True: + self.log.debug("Waiting for request") obj = self._workQueue.get() + self.log.debug("Popped request") if obj is None: break client, msg = obj + self.log.debug("Parsing message") self._parseMessage(client, msg) def _parseMessage(self, client, msg): @@ -560,4 +564,5 @@ self._threadFactory(partial(self._serveRequest, ctx, request)) def stop(self): + self.log.info("Stopping JsonRPC Server") self._workQueue.put_nowait(None) diff --git a/lib/yajsonrpc/betterAsyncore.py b/lib/yajsonrpc/betterAsyncore.py index 8806ee5..1e2c5ba 100644 --- a/lib/yajsonrpc/betterAsyncore.py +++ b/lib/yajsonrpc/betterAsyncore.py @@ -41,6 +41,9 @@ self.connection = connection self.__wctx = wrappedContext + def pending(self): + return self.connection.pending() + def get_context(self): return self.__wctx diff --git a/lib/yajsonrpc/stomp.py b/lib/yajsonrpc/stomp.py new file mode 100644 index 0000000..3e2ecfb --- /dev/null +++ b/lib/yajsonrpc/stomp.py @@ -0,0 +1,522 @@ +import logging +import socket +import cStringIO +from threading import Timer, Event +from uuid import uuid4 +from collections import deque +import time + +from betterAsyncore import Dispatcher +import asyncore + + +_ESCAPE_CHARS = (('\\\\', '\\'), ('\\r', '\r'), ('\\n', '\n'), ('\\c', ':')) + + +class Command: + MESSAGE = "MESSAGE" + SEND = "SEND" + SUBSCRIBE = "SUBSCRIBE" + UNSUBSCRIBE = "UNSUBSCRIBE" + CONNECT = "CONNECT" + CONNECTED = "CONNECTED" + ERROR = "ERROR" + RECEIPT = "RECEIPT" + + +class AckMode: + AUTO = "auto" + + +class StompError(RuntimeError): + def __init__(self, frame): + RuntimeError.__init__(self, frame.body) + + +class Frame(object): + def __init__(self, command="", headers=None, body=None): + self.command = command + if headers is None: + headers = {} + + self.headers = headers + if isinstance(body, unicode): + body = body.encode('utf-8') + + self.body = body + + def encode(self): + body = self.body + # We do it here so we are sure header is up to date + if body is not None: + self.headers["content-length"] = len(body) + + data = self.command + '\n' + data += '\n'.join(["%s:%s" % (encodeValue(key), encodeValue(value)) + for key, value in self.headers.iteritems()]) + data += '\n\n' + if body is not None: + data += body + + data += "\0" + return data + + def __repr__(self): + return "<StompFrame command=%s>" % (repr(self.command)) + + def copy(self): + return Frame(self.command, self.headers.copy(), self.body) + + +def decodeValue(s): + s = s.decode('utf-8') + for escaped, raw in _ESCAPE_CHARS: + s = s.replace(escaped, raw) + + # TODO : Throw error on invalid escape char + # spec: Undefined escape sequences such as \t (octet 92 and 116) MUST be + # treated as a fatal protocol error. + return s + + +def encodeValue(s): + if not isinstance(s, (str, unicode)): + s = str(s) + for escaped, raw in _ESCAPE_CHARS: + s = s.replace(raw, escaped) + + if isinstance(s, unicode): + s = s.encode('utf-8') + + return s + + +class Parser(object): + _STATE_CMD = "Parsing command" + _STATE_HEADER = "Parsing headers" + _STATE_BODY = "Receiving body" + + def __init__(self): + self._states = { + self._STATE_CMD: self._parse_command, + self._STATE_HEADER: self._parse_header, + self._STATE_BODY: self._parse_body} + self._frames = deque() + self._state = self._STATE_CMD + self._contentLength = -1 + self._flush() + + def _flush(self): + self._buffer = cStringIO.StringIO() + + def _handle_terminator(self, term): + if term not in self._buffer.getvalue(): + return None + + data = self._buffer.getvalue() + res, rest = data.split(term, 1) + self._flush() + self._buffer.write(rest) + + return res + + def _parse_command(self): + cmd = self._handle_terminator('\n') + if cmd is None: + return False + + if len(cmd) > 0 and cmd[-1] == '\r': + cmd = cmd[:-1] + + if cmd == "": + return True + + cmd = decodeValue(cmd) + self._tmpFrame = Frame(cmd) + + self._state = self._STATE_HEADER + return True + + def _parse_header(self): + header = self._handle_terminator('\n') + if header is None: + return False + + headers = self._tmpFrame.headers + if len(header) > 0 and header[-1] == '\r': + header = header[:-1] + + if header == "": + self._contentLength = int(headers.get('content-length', -1)) + self._state = self._STATE_BODY + return True + + # TODO: Proper error on missing or to many ':' + key, value = header.split(":") + key = decodeValue(key) + value = decodeValue(value) + + # If a client or a server receives repeated frame header entries, only + # the first header entry SHOULD be used as the value of header entry. + # Subsequent values are only used to maintain a history of state + # changes of the header and MAY be ignored. + if key not in headers: + headers[key] = value + + return True + + def _pushFrame(self): + self._frames.append(self._tmpFrame) + self._state = self._STATE_CMD + self._tmpFrame = None + self._contentLength = -1 + + def _parse_body(self): + if self._contentLength >= 0: + return self._parse_body_length() + else: + return self._parse_body_terminator() + + def _parse_body_terminator(self): + body = self._handle_terminator('\0') + if body is None: + return False + + self._tmpFrame.body = body + self._pushFrame() + return True + + def _parse_body_length(self): + buf = self._buffer + cl = self._contentLength + ndata = buf.tell() + if ndata < cl: + return False + + remainingBytes = 0 + self._flush() + body = buf.getvalue() + self._buffer.write(body[cl + 1:]) + body = body[:cl] + + if remainingBytes == 0: + self._tmpFrame.body = body + self._pushFrame() + + return True + + @property + def pending(self): + return len(self._frames) + + def parse(self, data): + states = self._states + self._buffer.write(data) + while states[self._state](): + pass + + def popFrame(self): + try: + return self._frames.popleft() + except IndexError: + return None + + +class Client(object): + def __init__(self, sock=None): + """ + Initialize the client. + + The socket parameter can be an already initialized socket. Should be + used to pass specialized socket objects like SSL sockets. + """ + if sock is None: + sock = self._sock = socket.socket() + else: + sock = sock + + self._map = {} + # Because we don't know how large the frames are + # we have to use non bolocking IO + sock.setblocking(False) + + # We have our own timeout for operations we + # pretend to be synchronous (like connect). + self._timeout = None + self._connected = Event() + self._subscriptions = {} + + self._aclient = None + self._adisp = None + + self._inbox = deque() + + @property + def outgoing(self): + return self._adisp.outgoing + + def _registerSubscription(self, sub): + self._subscriptions[sub.id] = sub + + def _unregisterSubscription(self, sub): + del self._subscriptions[sub.id] + + @property + def connected(self): + return self._connected.isSet() + + def handle_connect(self, aclient, frame): + self._connected.set() + + def handle_message(self, aclient, frame): + self._inbox.append(frame) + + def process(self, timeout=None): + if timeout is None: + timeout = self._timeout + + asyncore.loop(use_poll=True, timeout=timeout, map=self._map, count=1) + + def connect(self, address, hostname): + sock = self._sock + + self._aclient = AsyncClient(self, hostname) + adisp = self._adisp = AsyncDispatcher(self._aclient) + disp = self._disp = Dispatcher(adisp, sock, self._map) + sock.setblocking(True) + disp.connect(address) + sock.setblocking(False) + self.recv() # wait for CONNECTED msg + + if not self._connected.isSet(): + sock.close() + raise socket.error() + + def recv(self): + timeout = self._timeout + s = time.time() + duration = 0 + while timeout is None or duration < timeout: + try: + return self._inbox.popleft() + except IndexError: + td = timeout - duration if timeout is not None else None + self.process(td) + duration = time.time() - s + + return None + + def put_subscribe(self, destination, ack=None): + subid = self._aclient.subscribe(self._adisp, destination, ack) + sub = Subsciption(self, subid, ack) + self._registerSubscription(sub) + return sub + + def put_send(self, destination, data="", headers=None): + self._aclient.send(self._adisp, destination, data, headers) + + def put(self, frame): + self._adisp.send_raw(frame) + + def send(self): + disp = self._disp + timeout = self._timeout + duration = 0 + s = time.time() + while ((timeout is None or duration < timeout) and + (disp.writable() or not self._connected.isSet())): + td = timeout - duration if timeout is not None else None + self.process(td) + duration = time.time() - s + + def gettimout(self): + return self._timeout + + def settimeout(self, value): + self._timeout = value + + +class AsyncDispatcher(object): + log = logging.getLogger("stomp.AsyncDispatcher") + + def __init__(self, frameHandler, bufferSize=1024): + self._frameHandler = frameHandler + self._bufferSize = bufferSize + self._parser = Parser() + self._outbox = deque() + self._outbuf = None + + def _queueFrame(self, frame): + self._outbox.append(frame) + + @property + def outgoing(self): + n = len(self._outbox) + if self._outbuf != "": + n += 1 + + return n + + def handle_connect(self, dispatcher): + self._frameHandler.handle_connect(self) + + def handle_read(self, dispatcher): + pending = self._bufferSize + while pending: + try: + data = dispatcher.recv(pending) + except socket.error: + dispatcher.handle_error() + return + + try: + pending = dispatcher.socket.pending() + except AttributeError: + pending = 0 + pass + + parser = self._parser + + if data is not None: + parser.parse(data) + + frameHandler = self._frameHandler + if hasattr(frameHandler, "handle_frame"): + while parser.pending > 0: + frameHandler.handle_frame(self, parser.popFrame()) + + def popFrame(self): + return self._parser.popFrame() + + def handle_write(self, dispatcher): + if self._outbuf is None: + try: + frame = self._outbox.popleft() + except IndexError: + return + + self._outbuf = frame.encode() + + data = self._outbuf + numSent = dispatcher.send(data) + if numSent == len(data): + self._outbuf = None + else: + self._outbuf = data[numSent:] + + def send_raw(self, frame): + self._queueFrame(frame) + + def writable(self, dispatcher): + return len(self._outbox) > 0 or self._outbuf is not None + + def readable(self, dispatcher): + return True + + +class AsyncClient(object): + log = logging.getLogger("yajsonrpc.protocols.stomp.AsyncClient") + + def __init__(self, frameHandler, hostname): + self._hostname = hostname + self._frameHandler = frameHandler + self._connected = False + self._error = None + self._commands = { + Command.CONNECTED: self._process_connected, + Command.MESSAGE: self._process_message, + Command.RECEIPT: self._process_receipt, + Command.ERROR: self._process_error} + + @property + def connected(self): + return self._connected + + def getLastError(self): + return self._error + + def handle_connect(self, dispatcher): + self.log.debug("Sending CONNECT frame...") + hostname = self._hostname + frame = Frame( + Command.CONNECT, + {"accept-version": "1.2", + "host": hostname}) + + dispatcher.send_raw(frame) + + def handle_frame(self, dispatcher, frame): + self._commands[frame.command](frame, dispatcher) + + def _process_connected(self, frame, dispatcher): + self._connected = True + frameHandler = self._frameHandler + if hasattr(frameHandler, "handle_connect"): + frameHandler.handle_connect(self, frame) + + self.log.debug("Stomp connection astablished") + + def _process_message(self, frame, dispatcher): + frameHandler = self._frameHandler + + if hasattr(frameHandler, "handle_message"): + frameHandler.handle_message(self, frame) + + def _process_receipt(self, frame, dispatcher): + # TODO: + pass + + def _process_error(self, frame, dispatcher): + raise StompError(frame) + + def send(self, dispatcher, destination, data="", headers=None): + frame = Frame( + Command.SEND, + {"destination": destination}, + data) + + dispatcher.send_raw(frame) + + def subscribe(self, dispatcher, destination, ack=None): + if ack is None: + ack = AckMode.AUTO + + subscriptionID = str(uuid4()) + + frame = Frame( + Command.SUBSCRIBE, + {"destination": destination, + "ack": ack, + "id": subscriptionID}) + + dispatcher.send_raw(frame) + + return subscriptionID + + +class Subsciption(object): + def __init__(self, client, subid, ack): + self._ack = ack + self._subid = subid + self._client = client + self._valid = True + + @property + def id(self): + return self._subid + + def unsubscribe(self): + client = self._client + subid = self._subid + + client._unregisterSubscription(self) + + frame = Frame(Command.UNSUBSCRIBE, + {"id": str(subid)}) + client.put(frame) + self._valid = False + + def __del__(self): + # Using a timer because unsubscribe action might involve taking locks. + if self._valid: + Timer(0, self.unsubscribe) diff --git a/lib/yajsonrpc/stompReactor.py b/lib/yajsonrpc/stompReactor.py new file mode 100644 index 0000000..7448d76 --- /dev/null +++ b/lib/yajsonrpc/stompReactor.py @@ -0,0 +1,335 @@ +# Copyright (C) 2014 Saggi Mizrahi, 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 version 2 as +# published by the Free Software Foundation. +# +# 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 St, Fifth Floor, Boston, MA 02110-1301 USA + +import struct +import asyncore +import socket +import os +import threading +import logging + +import stomp + +from betterAsyncore import \ + Dispatcher, \ + SSLDispatcher, \ + SSLContext + + +_Size = struct.Struct("!Q") + +_STATE_LEN = "Waiting for message length" +_STATE_MSG = "Waiting for message" + + +_DEFAULT_RESPONSE_DESTINATIOM = "/queue/_local/vdsm/reponses" +_DEFAULT_REQUEST_DESTINATION = "/queue/_local/vdsm/requests" + + +def _SSLContextFactory(ctxdef): + """Creates an appropriate ssl context from the generic defenition defined + in __init__.py""" + return SSLContext(cert_file=ctxdef.cert_file, + key_file=ctxdef.key_file, + ca_cert=ctxdef.ca_cert, + session_id=ctxdef.session_id, + protocol=ctxdef.protocol) + + +class StompAdapterImpl(object): + log = logging.getLogger("Broker.StompAdapter") + + def __init__(self, reactor, messageHandler): + self._reactor = reactor + self._messageHandler = messageHandler + self._commands = { + stomp.Command.CONNECT: self._cmd_connect, + stomp.Command.SEND: self._cmd_send, + stomp.Command.SUBSCRIBE: self._cmd_subscribe, + stomp.Command.UNSUBSCRIBE: self._cmd_unsubscribe} + + def _cmd_connect(self, dispatcher, frame): + self.log.info("Processing CONNECT request") + version = frame.headers.get("accept-version", None) + if version != "1.2": + res = stomp.Frame(stomp.Command.ERROR, None, "Version unsupported") + + else: + res = stomp.Frame(stomp.Command.CONNECTED, {"version": "1.2"}) + + dispatcher.send_raw(res) + self.log.info("CONNECT response queued") + self._reactor.wakeup() + + def _cmd_subscribe(self, dispatcher, frame): + self.log.debug("Subscribe command ignored") + + def _cmd_unsubscribe(self, dispatcher, frame): + self.log.debug("Unsubscribe command ignored") + + def _cmd_send(self, dispatcher, frame): + self.log.debug("Passing incoming message") + self._messageHandler(self, frame.body) + + def handle_frame(self, dispatcher, frame): + self.log.debug("Handling message %s", frame) + try: + self._commands[frame.command](dispatcher, frame) + except KeyError: + self.log.warn("Unknown command %s", frame) + dispatcher.handle_error() + + +class StompServer(object): + log = logging.getLogger("yajsonrpc.StompServer") + + def __init__(self, sock, reactor, addr, sslctx=None): + self._addr = addr + self._reactor = reactor + self._messageHandler = None + + adapter = StompAdapterImpl(reactor, self._handleMessage) + adisp = self._adisp = stomp.AsyncDispatcher(adapter) + if sslctx is None: + try: + sslctx = sock.get_context() + except AttributeError: + # if get_context() is missing it just means we recieved an + # socket that doesn't use SSL + pass + else: + sslctx = _SSLContextFactory(sslctx) + + if sslctx is None: + dispatcher = Dispatcher(adisp, sock=sock, map=reactor._map) + else: + dispatcher = SSLDispatcher(adisp, sslctx, sock=sock, + map=reactor._map) + + if sock is None: + address_family = socket.getaddrinfo(*addr)[0][0] + dispatcher.create_socket(address_family, socket.SOCK_STREAM) + + self._dispatcher = dispatcher + + def setTimeout(self, timeout): + self._dispatcher.socket.settimeout(timeout) + + def connect(self): + self._dispatcher.connect(self._addr) + + def _handleMessage(self, impl, data): + if self._messageHandler is not None: + self.log.debug("Processing incoming message") + self._messageHandler((self, data)) + + def setMessageHandler(self, msgHandler): + self._messageHandler = msgHandler + + def send(self, message): + self.log.debug("Sending response") + res = stomp.Frame(stomp.Command.MESSAGE, + {"destination": _DEFAULT_RESPONSE_DESTINATIOM, + "content-type": "application/json"}, + message) + self._adisp.send_raw(res) + self._reactor.wakeup() + + def close(self): + self._dispatcher.close() + + +class StompClient(object): + log = logging.getLogger("jsonrpc.AsyncoreClient") + + def __init__(self, sock, reactor, addr, sslctx=None): + self._addr = addr + self._reactor = reactor + self._messageHandler = None + + self._aclient = stomp.AsyncClient(self, "vdsm") + adisp = self._adisp = stomp.AsyncDispatcher(self._aclient) + if sslctx is None: + try: + sslctx = sock.get_context() + except AttributeError: + # if get_context() is missing it just means we recieved an + # socket that doesn't use SSL + pass + else: + sslctx = _SSLContextFactory(sslctx) + + if sslctx is None: + dispatcher = Dispatcher(adisp, sock=sock, map=reactor._map) + else: + dispatcher = SSLDispatcher(adisp, sslctx, sock=sock, + map=reactor._map) + + if sock is None: + address_family = socket.getaddrinfo(*addr)[0][0] + dispatcher.create_socket(address_family, socket.SOCK_STREAM) + + self._dispatcher = dispatcher + + def setTimeout(self, timeout): + self._dispatcher.socket.settimeout(timeout) + + def connect(self): + self._dispatcher.connect(self._addr) + + def handle_message(self, impl, frame): + if self._messageHandler is not None: + self.log.debug("Queueing incoming message") + self._messageHandler((self, frame.body)) + + def setMessageHandler(self, msgHandler): + self._messageHandler = msgHandler + + def send(self, message): + self._aclient.send(self._adisp, _DEFAULT_REQUEST_DESTINATION, + message, + {"content-type": "application/json"}) + self._reactor.wakeup() + self.log.debug("Message queued for delivery") + + def close(self): + self._dispatcher.close() + + +def StompListener(reactor, address, acceptHandler, sslctx=None): + impl = StompListenerImpl(reactor, address, acceptHandler) + if sslctx is None: + return Dispatcher(impl, map=reactor._map) + else: + sslctx = _SSLContextFactory(sslctx) + return SSLDispatcher(impl, sslctx, map=reactor._map) + + +# FIXME: We should go about making a listener wrapper like the client wrapper +# This is not as high priority as users don't interact with listeners +# as much +class StompListenerImpl(object): + log = logging.getLogger("jsonrpc.StompListener") + + def __init__(self, reactor, address, acceptHandler): + self._reactor = reactor + self._address = address + self._acceptHandler = acceptHandler + + def init(self, dispatcher): + address_family = socket.getaddrinfo(*self._address)[0][0] + dispatcher.create_socket(address_family, socket.SOCK_STREAM) + + dispatcher.set_reuse_addr() + dispatcher.bind(self._address) + dispatcher.listen(5) + + def handle_accept(self, dispatcher): + try: + pair = dispatcher.accept() + except Exception as e: + self.log.exception(e) + raise + if pair is None: + return + + sock, addr = pair + self.log.debug("Accepting connection from client " + "at tcp://%s:%s", addr[0], addr[1]) + + client = StompServer(sock, self._reactor, addr) + self._acceptHandler(self, client) + + def writable(self, dispatcher): + return False + + +class _AsyncoreEvent(asyncore.file_dispatcher): + def __init__(self, map): + self._lock = threading.Lock() + r, w = os.pipe() + self._w = w + try: + asyncore.file_dispatcher.__init__(self, r, map=map) + except: + os.close(r) + os.close(w) + raise + + # the file_dispatcher ctor dups the file in order to take ownership of + # it + os.close(r) + self._isSet = False + + def writable(self): + return False + + def set(self): + with self._lock: + if self._isSet: + return + + self._isSet = True + + os.write(self._w, "a") + + def handle_read(self): + with self._lock: + self.recv(1) + self._isSet = False + + def close(self): + try: + os.close(self._w) + except (OSError, IOError): + pass + + asyncore.file_dispatcher.close(self) + + +class StompReactor(object): + def __init__(self, sslctx=None): + self.sslctx = sslctx + self._map = {} + self._isRunning = False + self._wakeupEvent = _AsyncoreEvent(self._map) + + def createListener(self, address, acceptHandler): + listener = StompListener(self, address, acceptHandler, self.sslctx) + self.wakeup() + return listener + + def createClient(self, address): + return StompClient(None, self, address, self.sslctx) + + def process_requests(self): + self._isRunning = True + while self._isRunning: + asyncore.loop(use_poll=True, map=self._map, count=1) + + for key, dispatcher in self._map.items(): + del self._map[key] + dispatcher.close() + + def wakeup(self): + self._wakeupEvent.set() + + def stop(self): + self._isRunning = False + try: + self.wakeup() + except (IOError, OSError): + # Client woke up and closed the event dispatcher without our help + pass diff --git a/tests/jsonRpcTests.py b/tests/jsonRpcTests.py index 3b63c01..3eb210b 100644 --- a/tests/jsonRpcTests.py +++ b/tests/jsonRpcTests.py @@ -149,7 +149,10 @@ class _DummyBridge(object): + log = logging.getLogger("tests.DummyBridge") + def echo(self, text): + self.log.info("ECHO: '%s'", text) return text def ping(self): @@ -185,6 +188,7 @@ bridge = _DummyBridge() with constructServer(rt, bridge, ssl) as (server, clientFactory): with self._client(clientFactory) as client: + self.log.info("Calling 'echo'") self.assertEquals(self._callTimeout(client, "echo", (data,), apiTests.id, CALL_TIMEOUT), data) diff --git a/tests/jsonRpcUtils.py b/tests/jsonRpcUtils.py index 6d6dcf2..e3f54f6 100644 --- a/tests/jsonRpcUtils.py +++ b/tests/jsonRpcUtils.py @@ -10,6 +10,7 @@ from yajsonrpc import \ JsonRpcServer, \ asyncoreReactor, \ + stompReactor, \ JsonRpcClientPool, \ SSLContext @@ -36,6 +37,14 @@ @contextmanager +def _stompServerConstructor(): + port = getFreePort() + address = ("127.0.0.1", port) + reactorType = stompReactor.StompReactor + yield reactorType, address + + +@contextmanager def _tcpServerConstructor(): port = getFreePort() address = ("127.0.0.1", port) @@ -56,7 +65,8 @@ REACTOR_CONSTRUCTORS = {"tcp": _tcpServerConstructor, - "amqp": _protonServerConstructor} + "amqp": _protonServerConstructor, + "stomp": _stompServerConstructor} REACTOR_TYPE_PERMUTATIONS = [[r] for r in REACTOR_CONSTRUCTORS.iterkeys()] CONNECTION_PERMUTATIONS = tuple(product(REACTOR_CONSTRUCTORS.iterkeys(), (True, False))) diff --git a/vdsm_api/BindingJsonRpc.py b/vdsm_api/BindingJsonRpc.py index 6c0fa47..471cb70 100644 --- a/vdsm_api/BindingJsonRpc.py +++ b/vdsm_api/BindingJsonRpc.py @@ -21,6 +21,7 @@ from yajsonrpc import JsonRpcServer from yajsonrpc.asyncoreReactor import AsyncoreReactor +from yajsonrpc.stompReactor import StompReactor from yajsonrpc.betterAsyncore import SSLContext ProtonReactor = None try: @@ -49,6 +50,9 @@ if backendType not in reactors: if backendType == "tcp": reactors["tcp"] = self._createTcpReactor(truststore_path) + elif backendType == "stomp": + reactors["stomp"] = \ + self._createStompReactor(truststore_path) elif backendType == "amqp": if ProtonReactor is None: continue @@ -84,6 +88,15 @@ ca_cert = truststore_path + '/certs/cacert.pem' return AsyncoreReactor(SSLContext(cert_file, key_file, ca_cert)) + def _createStompReactor(self, truststore_path=None): + if truststore_path is None: + return StompReactor() + else: + key_file = truststore_path + '/keys/vdsmkey.pem' + cert_file = truststore_path + '/certs/vdsmcert.pem' + ca_cert = truststore_path + '/certs/cacert.pem' + return StompReactor(SSLContext(cert_file, key_file, ca_cert)) + def _createProtonReactor(self): return ProtonReactor() @@ -104,6 +117,8 @@ try: if backendType == "tcp": self._createTcpListener(cfg) + if backendType == "stomp": + self._createStompListener(cfg) elif backendType == "amqp": self._createProtonListener(cfg) except: -- To view, visit http://gerrit.ovirt.org/26750 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: I22bcae1e150dea7bc7d9fecefb6847c48bfe8949 Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Saggi Mizrahi <smizrahi(a)redhat.com>

4 31

Change in vdsm[master]: json-rpc: Protocol detection
by Nir Soffer 05 Jun '14

05 Jun '14

Nir Soffer has posted comments on this change. Change subject: json-rpc: Protocol detection ...................................................................... Patch Set 32: (1 comment) http://gerrit.ovirt.org/#/c/26300/32/lib/yajsonrpc/__init__.py File lib/yajsonrpc/__init__.py: Line 488: res = True if res is None else res Line 489: Line 490: ctx.requestDone(JsonRpcResponse(res, None, req.id)) Line 491: Line 492: @traceback(on=log.name) > Nir suggested to use it. I was suggesting this on the protocol detector, not here, because I did not review this file. This decorator makes code simpler because you don't need to add try blocks around the body of threads target function, and it will not break when your error handling code is broken. This does not replace proper error handling in the thread, but a last resort error handler, ensuring that when the thread dies in the field (because error handling was broken), it will be logged. So instead of: def mainloop(self): try: do stuff and other stuff even more except Exception: logging.exception("thread foo died") You can do: @traceback(on=log.name, msg="thread foo died") def mainloop(self): do stuff and other stuff even more Looking in function body, there is no error handling in the while loop, so any unexpected failure in one of statements will cause the thread to abort silently, which makes it impossible to support this code in the field. Line 493: def serve_requests(self): Line 494: while True: Line 495: self.log.debug("Waiting for request") Line 496: obj = self._workQueue.get() -- To view, visit http://gerrit.ovirt.org/26300 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: comment Gerrit-Change-Id: Id739a40e2b37dcc175137ec91cd5ec166ad24a75 Gerrit-PatchSet: 32 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Piotr Kliczewski <piotr.kliczewski(a)gmail.com> Gerrit-Reviewer: Barak Azulay <bazulay(a)redhat.com> Gerrit-Reviewer: Francesco Romani <fromani(a)redhat.com> Gerrit-Reviewer: Liron Ar <laravot(a)redhat.com> Gerrit-Reviewer: Michal Skrivanek <michal.skrivanek(a)redhat.com> Gerrit-Reviewer: Nir Soffer <nsoffer(a)redhat.com> Gerrit-Reviewer: Piotr Kliczewski <piotr.kliczewski(a)gmail.com> Gerrit-Reviewer: Saggi Mizrahi <smizrahi(a)redhat.com> Gerrit-Reviewer: Vinzenz Feenstra <vfeenstr(a)redhat.com> Gerrit-Reviewer: Yaniv Bronhaim <ybronhei(a)redhat.com> Gerrit-Reviewer: Yeela Kaplan <ykaplan(a)redhat.com> Gerrit-Reviewer: automation(a)ovirt.org Gerrit-Reviewer: oVirt Jenkins CI Server Gerrit-HasComments: Yes

1 0

Jump to page:

2025

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

vdsm-patches June 2014