Change in vdsm[master]: introducing downloadImageFromStream

laravot at redhat.com laravot at redhat.com
Wed Jan 15 14:43:55 UTC 2014


Liron Ar has uploaded a new change for review.

Change subject: introducing downloadImageFromStream
......................................................................

introducing downloadImageFromStream

This patch introduces the downloadImageFromStream verb in vdsm which
allows to upload using streaming content to vdsm images.
XML-RPC spec doesn't support streaming, as we don't want to use HTTP
server in addition to the current XML-RPC server (to not use another
port) we will threat some requests as http requests instead of xml-rpc
requests.

General upload information:
---------------------------------------------------------------
PUT requests arriving to the server with content type of
application/octet-stream to default paths that we use
today for request handling ('/', '/RPC2') will be threated as
upload requests.

The PUT stream upload request inspected headers are:
-- Mandatory headers:
        *content-length

-- Mandatory custom headers:
        *X-Storage-Pool-Id : mandatory for succesfull execution
        *X-Storage-Domain-Id: mandatory for succesfull execution
        *X-Image-Id: mandatory for succesfull execution
        *X-Volume-Id: mandatory for succesfull execution

-- Optional custom headers;
        *X-Buffer-Size: optional, indicates the passed buffer

The server will return answers with the following codes:
200 - request completed, the response body will contain the execution
results, the returned content type is 'application/json'
500 - internal error
406 - missing content-length header

in case that a task was created, its id will be passed in the response
using the custom header 'X-Task-Id' for polling by the engine.

Implementation details
--------------------------------------------------------------
Currently for any spm task (besides delete image) the operation is
performed through a spm task.
We need to have a task to indicate that "there's something" going on -
for example, if we had no task during the upload the spm could be
stopped and be started on other host which might be hazardous.

For having a task we can do two different things -
1. schedule a task to perform the whole download process, that means that
the request thread will be blocked, the read from the stream and writes
to the destination image will be done through the task and then the
request thread will be released and response would be sent.

2. schedule a "dummy" task with "blank" function that will just be used to
indicate that there's an downloadFromStream execution going on while the
request thread will be used for executing the actual download.

Option 2 is implemented with this patch.

needed:
*socket timeout inspection
*tests?

Change-Id: I768b84799ed9fb2769c6d4240519d036f8988b99
Signed-off-by: Liron Aravot <laravot at redhat.com>
---
M vdsm/API.py
M vdsm/BindingXMLRPC.py
M vdsm/storage/hsm.py
M vdsm/storage/imageSharing.py
4 files changed, 153 insertions(+), 3 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/81/23281/1

diff --git a/vdsm/API.py b/vdsm/API.py
index e7a550c..37802e1 100644
--- a/vdsm/API.py
+++ b/vdsm/API.py
@@ -838,6 +838,10 @@
         return self._irs.downloadImage(
             methodArgs, self._spUUID, self._sdUUID, self._UUID, volUUID)
 
+    def downloadFromStream(self, methodArgs, volUUID=None):
+        return self._irs.downloadImageFromStream(
+            methodArgs, self._spUUID, self._sdUUID, self._UUID, volUUID)
+
 
 class LVMVolumeGroup(APIBase):
     ctorArgs = ['lvmvolumegroupID']
diff --git a/vdsm/BindingXMLRPC.py b/vdsm/BindingXMLRPC.py
index a340ae7..8b7d9fa 100644
--- a/vdsm/BindingXMLRPC.py
+++ b/vdsm/BindingXMLRPC.py
@@ -25,11 +25,14 @@
 import logging
 import libvirt
 import threading
+import cgi
+import json
 
 from vdsm import constants
 from vdsm import utils
 from vdsm.define import doneCode, errCode
 from vdsm.netinfo import getRouteDeviceTo
+from storage.threadLocal import vars
 import API
 from vdsm.exception import VdsmException
 try:
@@ -128,10 +131,91 @@
         else:
             basehandler = SimpleXMLRPCServer.SimpleXMLRPCRequestHandler
 
-        class LoggingHandler(LoggingMixIn, basehandler):
+        class RequestHandler(LoggingMixIn, basehandler):
+
+            log = logging.getLogger("BindingXMLRPC.RequestHandler")
+            CUSTOM_HEADER_POOL = 'X-Storage-Pool-Id'
+            CUSTOM_HEADER_DOMAIN = 'X-Storage-Domain-Id'
+            CUSTOM_HEADER_IMAGE = 'X-Image-Id'
+            CUSTOM_HEADER_VOLUME = 'X-Volume-Id'
+            CUSTOM_HEADER_BUFFER = 'X-Buffer-Size'
+            CUSTOM_HEADER_TASK_ID = 'X-Task-Id'
+            HEADER_CONTENT_LENGTH = 'content-length'
+            HEADER_CONTENT_TYPE = 'content-type'
+            RESPONSE_LENGTH_REQUIRED_ = 411
+            RESPONSE_NOT_ACCEPTABLE = 406
+            RESPONSE_REQUEST_SUCCEEDED = 200
+            RESPONSE_INTERNAL_ERROR = 500
+
             def setup(self):
                 threadLocal.client = self.client_address[0]
                 return basehandler.setup(self)
+
+            def addTaskIdHeaderIfPresent(self):
+                if hasattr(vars, 'task'):
+                    self.send_header(self.CUSTOM_HEADER_TASK_ID,
+                                     vars.task.getID())
+
+            def do_PUT(self):
+                ctype, pt = cgi.parse_header(self.headers.getheader(self.
+                                             HEADER_CONTENT_TYPE))
+                try:
+                    if ctype == 'application/octet-stream':
+                        contentLength = self.headers. \
+                            getheader(self.HEADER_CONTENT_LENGTH)
+                        if not contentLength:
+                            self.send_response(411)
+                            self.finish()
+                            return
+
+                        # NEED TO DECIDE ABOUT THOSE
+                        #self.rfile._sock.settimeout(5)
+                        #self.rfile._sock.set_socket_read_timeout
+                        spUUID = self.headers. \
+                            getheader(self.CUSTOM_HEADER_POOL)
+                        sdUUID = self.headers. \
+                            getheader(self.CUSTOM_HEADER_DOMAIN)
+                        imgUUID = self.headers. \
+                            getheader(self.CUSTOM_HEADER_IMAGE)
+                        volUUID = self.headers. \
+                            getheader(self.CUSTOM_HEADER_VOLUME)
+                        bufferSize = self.headers. \
+                            getheader(self.CUSTOM_HEADER_BUFFER)
+                        contentLength = self.headers. \
+                            getheader(self.HEADER_CONTENT_LENGTH)
+
+                        methodArgs = {'method': 'stream',
+                                      'fileObj': self.rfile,
+                                      'Content-Length': contentLength,
+                                      'Buffer-Size': bufferSize}
+                        image = API.Image(imgUUID, spUUID, sdUUID)
+                        response = image.downloadFromStream(methodArgs,
+                                                            volUUID)
+                        json_response = json.dumps(response)
+                        self.send_response(200)
+                        self.send_header(self.HEADER_CONTENT_TYPE,
+                                         'application/json')
+                        self.send_header(self.HEADER_CONTENT_LENGTH,
+                                         len(json_response))
+                        self.addTaskIdHeaderIfPresent()
+                        self.end_headers()
+                        self.wfile.write(json_response)
+                        self.finish()
+                    else:
+                        self.send_response(406)
+                        self.finish()
+                # shouldn't happen
+                except Exception:
+                    self.log.error("execution exception",
+                                   exc_info=True)
+                    try:
+                        self.send_response(500)
+                        self.addTaskIdHeaderIfPresent()
+                        self.end_headers()
+                        self.finish()
+                    except Exception:
+                        self.log.error("failed to return response",
+                                       exc_info=True)
 
             def parse_request(self):
                 r = (SecureXMLRPCServer.SecureXMLRPCRequestHandler.
@@ -145,11 +229,11 @@
                 server_address,
                 keyfile=KEYFILE, certfile=CERTFILE, ca_certs=CACERT,
                 timeout=self.serverRespTimeout,
-                requestHandler=LoggingHandler)
+                requestHandler=RequestHandler)
         else:
             server = utils.SimpleThreadedXMLRPCServer(
                 server_address,
-                requestHandler=LoggingHandler, logRequests=True)
+                requestHandler=RequestHandler, logRequests=True)
         utils.closeOnExec(server.socket.fileno())
 
         server.lastClientTime = 0
diff --git a/vdsm/storage/hsm.py b/vdsm/storage/hsm.py
index 31a6e1c..42b54ce 100644
--- a/vdsm/storage/hsm.py
+++ b/vdsm/storage/hsm.py
@@ -1674,6 +1674,18 @@
         self._spmSchedule(spUUID, "downloadImage", pool.downloadImage,
                           methodArgs, sdUUID, imgUUID, volUUID)
 
+    @public
+    def downloadImageFromStream(self, methodArgs, spUUID, sdUUID, imgUUID,
+                                volUUID=None):
+        """
+        Download an image from a stream synchronously.
+        """
+        sdCache.produce(sdUUID)
+        pool = self.getPool(spUUID)
+        self._spmSchedule(spUUID, "downloadImageFromStream %s" %
+                          imgUUID, lambda: True)
+        pool.downloadImage(methodArgs, sdUUID, imgUUID, volUUID)
+
     @deprecated
     @public
     def moveMultipleImages(self, spUUID, srcDomUUID, dstDomUUID, imgDict,
diff --git a/vdsm/storage/imageSharing.py b/vdsm/storage/imageSharing.py
index 26f299a..fa2080d 100644
--- a/vdsm/storage/imageSharing.py
+++ b/vdsm/storage/imageSharing.py
@@ -20,6 +20,9 @@
 import logging
 
 import curlImgWrap
+import storage_exception as se
+from vdsm import constants
+from vdsm import utils
 
 
 log = logging.getLogger("Storage.ImageSharing")
@@ -46,6 +49,20 @@
     return size
 
 
+def streamGetSize(methodArgs):
+    size = None
+    if 'Content-Length' in methodArgs:
+        size = int(methodArgs['Content-Length'])
+    return size
+
+
+def streamGetBufferSize(methodArgs):
+    size = None
+    if 'X-Buffer-Size' in methodArgs:
+        size = int(methodArgs['X-Buffer-Size'])
+    return size
+
+
 def httpDownloadImage(dstImgPath, methodArgs):
     curlImgWrap.download(methodArgs.get('url'), dstImgPath,
                          methodArgs.get("headers", {}))
@@ -56,8 +73,41 @@
                        methodArgs.get("headers", {}))
 
 
+def streamDownloadImage(dstImgPath, methodArgs):
+    bytes_left = streamGetSize(methodArgs)
+    buffer_size = streamGetBufferSize(methodArgs)
+    stream = methodArgs.get('fileObj')
+
+    BS = constants.MEGAB
+    count = bytes_left / BS
+    cmd = [constants.EXT_DD, "of=%s" % dstImgPath, "bs=%s" %
+           constants.MEGAB, "count=%d" % count]
+    p = utils.execCmd(cmd, sudo=False, sync=False)
+
+    default_read_size = buffer_size
+    if default_read_size is None:
+        default_read_size = BS
+
+    while bytes_left > 0:
+        i = min(default_read_size, bytes_left)
+        data = stream.read(i)
+        if not data:
+            break
+        p.stdin.write(data)
+        p.stdin.flush()
+        bytes_left = bytes_left - i
+
+    p.stdin.close()
+    p.wait()
+    if p.returncode != 0:
+        log.error("streamDownloadImage failed",
+                  p.returncode, p.stderr.read(1000))
+        raise se.MiscFileWriteException()
+
+
 _METHOD_IMPLEMENTATIONS = {
     'http': (httpGetSize, httpDownloadImage, httpUploadImage),
+    'stream': (streamGetSize, streamDownloadImage, None),
 }
 
 


-- 
To view, visit http://gerrit.ovirt.org/23281
To unsubscribe, visit http://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: I768b84799ed9fb2769c6d4240519d036f8988b99
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Liron Ar <laravot at redhat.com>


More information about the vdsm-patches mailing list