On 22/04/2015 05:02, lvroyce@linux.vnet.ibm.com wrote:

From: Royce Lv <lvroyce@linux.vnet.ibm.com>

Volume upload will use the same REST api as 'capacity' type when creating storage volume. Following data transfer is implemented by several storage volume update: POST /storagepools/<pool-name>/storagevolumes/ {"capacity": 1000000, "format": "raw", "name": "volume-1"} PUT /storagepools/<pool-name>/storagevolumes/volume-1 {"index": "0", "chunk_size": "1024", "chunk": form-data} PUT /storagepools/<pool-name>/storagevolumes/volume-1 {"index": "1", "chunk_size": "1024", "chunk": form-data} ... PUT /storagepools/<pool-name>/storagevolumes/volume-1 {"index": "1024", "chunk_size": "1024", "chunk": form-data}

Signed-off-by: Royce Lv <lvroyce@linux.vnet.ibm.com> --- docs/API.md | 7 +++++-- src/kimchi/API.json | 22 ++++++++++++++++++++++ src/kimchi/i18n.py | 1 + 3 files changed, 28 insertions(+), 2 deletions(-)

diff --git a/docs/API.md b/docs/API.md index 3f7925f..2e71989 100644 --- a/docs/API.md +++ b/docs/API.md @@ -474,14 +474,13 @@ A interface represents available network interface on VM. in the defined Storage Pool * **POST**: Create a new Storage Volume in the Storage Pool The return resource is a task resource * See Resource: Task * - Only one of 'file', 'capacity', 'url' can be specified. + Only one of 'capacity', 'url' can be specified. * name: The name of the Storage Volume * capacity: The total space which can be used to store volumes The unit is bytes * format: The format of the defined Storage Volume. Only used when creating a storage volume with 'capacity'. * file: File to be uploaded, passed through form data - * url: URL to be downloaded

### Resource: Storage Volume

@@ -508,6 +507,10 @@ A interface represents available network interface on VM.

* **DELETE**: Remove the Storage Volume * **POST**: *See Storage Volume Actions* +* **PUT**: Upload storage volume chunk + * index: Chunk index of the slice in file. + * chunk_size: Chunk size of the slice in Bytes. + * chunk: Actual data of uploaded file

**Actions (POST):**

diff --git a/src/kimchi/API.json b/src/kimchi/API.json index 474661c..9df1dc0 100644 --- a/src/kimchi/API.json +++ b/src/kimchi/API.json @@ -222,6 +222,28 @@ } } }, + "storagevolume_update": { + "type": "object", + "properties": { + "chunk": { + "description": "Upload storage volume chunk", + "required": true + },

'error' message is missing for 'chunk' parameter.

+ "index": { + "description": "Chunk index of uploaded storage volume", + "type": "string", + "error": "KCHVOL0024E", + "required": true + }, + "chunk_size": { + "description": "Chunk size of uploaded storage volume", + "type": "string", + "error": "KCHVOL0024E", + "required": true + } + }, + "additionalProperties": false + }, "vms_create": { "type": "object", "error": "KCHVM0016E", diff --git a/src/kimchi/i18n.py b/src/kimchi/i18n.py index c8986cf..c012a62 100644 --- a/src/kimchi/i18n.py +++ b/src/kimchi/i18n.py @@ -215,6 +215,7 @@ messages = { "KCHVOL0021E": _("Storage volume URL must be http://, https://, ftp:// or ftps://."), "KCHVOL0022E": _("Unable to access file %(url)s. Please, check it."), "KCHVOL0023E": _("Unable to clone storage volume '%(name)s' in pool '%(pool)s'. Details: %(err)s"),

+ "KCHVOL0024E": _("Upload volume chunk index, size and total size must be integer"),

'total size' is not in used for storage volume update. You should have the message according to it.

"KCHIFACE0001E": _("Interface %(name)s does not exist"),

Reviewed-by: Aline Manera <alinefm@linux.vnet.ibm.com> On 22/04/2015 05:02, lvroyce@linux.vnet.ibm.com wrote:

From: Royce Lv <lvroyce@linux.vnet.ibm.com>

When update does not accept params in base class, cherrypy will raise error that extra params are provided in body. So allow update function to accept params.

Signed-off-by: Royce Lv <lvroyce@linux.vnet.ibm.com> --- src/kimchi/control/base.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/kimchi/control/base.py b/src/kimchi/control/base.py index b50ea5c..b9520c0 100644 --- a/src/kimchi/control/base.py +++ b/src/kimchi/control/base.py @@ -156,7 +156,7 @@ class Resource(object): raise cherrypy.HTTPError(400, e.message)

@cherrypy.expose - def index(self): + def index(self, *args, **kargs): method = validate_method(('GET', 'DELETE', 'PUT'), self.role_key, self.admin_methods)

@@ -167,7 +167,7 @@ class Resource(object):

return {'GET': self.get, 'DELETE': self.delete, - 'PUT': self.update}[method]() + 'PUT': self.update}[method](*args, **kargs) except InvalidOperation, e: raise cherrypy.HTTPError(400, e.message) except InvalidParameter, e: @@ -194,7 +194,7 @@ class Resource(object):

return user_name in users or len(set(user_groups) & set(groups)) > 0

- def update(self): + def update(self, *args, **kargs): try: update = getattr(self.model, model_fn(self, 'update')) except AttributeError:

On 22/04/2015 05:02, lvroyce@linux.vnet.ibm.com wrote:

From: Royce Lv <lvroyce@linux.vnet.ibm.com>

This lock facility guarentees 5 concurrent volume upload and make sure locks are created and reclaimed when needed.

Signed-off-by: Royce Lv <lvroyce@linux.vnet.ibm.com> --- src/kimchi/i18n.py | 1 + src/kimchi/model/utils.py | 46 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+)

diff --git a/src/kimchi/i18n.py b/src/kimchi/i18n.py index c012a62..b712c16 100644 --- a/src/kimchi/i18n.py +++ b/src/kimchi/i18n.py @@ -216,6 +216,7 @@ messages = { "KCHVOL0022E": _("Unable to access file %(url)s. Please, check it."), "KCHVOL0023E": _("Unable to clone storage volume '%(name)s' in pool '%(pool)s'. Details: %(err)s"), "KCHVOL0024E": _("Upload volume chunk index, size and total size must be integer"), + "KCHVOL0026E": _("Inconsistent upload count"),

A more detailed message would be useful to we really know what is going wrong.

"KCHIFACE0001E": _("Interface %(name)s does not exist"),

diff --git a/src/kimchi/model/utils.py b/src/kimchi/model/utils.py index b2739b2..b089be5 100644 --- a/src/kimchi/model/utils.py +++ b/src/kimchi/model/utils.py

The code below is only related to storage volume upload feature so I suggest to move it to model/storagevolumes.py The model/utils.py should be used only for generic matters.

@@ -19,6 +19,7 @@

import libvirt import socket +import threading import urlparse from lxml import etree, objectify from lxml.builder import E, ElementMaker @@ -28,6 +29,11 @@ from kimchi.model.featuretests import FeatureTests

KIMCHI_META_URL = "https://github.com/kimchi-project/kimchi" KIMCHI_NAMESPACE = "kimchi" +UPLOAD_THREADS = 5 # Concurrent upload volume counts at the same time + +upload_semaphore = threading.BoundedSemaphore(value=UPLOAD_THREADS) +upload_lock_pool = dict() +pool_lock = threading.Lock()

def get_vm_name(vm_name, t_name, name_list): @@ -162,3 +168,43 @@ def get_metadata_node(dom, tag, metadata_support, mode="current"): if node is not None: return etree.tostring(node) return "" + + +class UpdateLock(object): + def __init__(self): + self.ref_cnt = 0 + self.lock = threading.Lock() + + def get_lock(self): + self.ref_cnt += 1 + return self.lock + + def release_lock(self): + delete = False + self.ref_cnt -= 1 + if (self.ref_cnt == 0): + delete = True + return delete + + +def get_vol_update_lock(vol_path): + # upload_semaphore controls the max upload count + upload_semaphore.acquire() + + # pool lock make sure lock list get/store action is atomic + with pool_lock: + vol_lock = upload_lock_pool.get(vol_path) + if vol_lock: + return vol_lock.get_lock() + if len(upload_lock_pool.keys()) > (UPLOAD_THREADS - 1): + raise OperationFailed("KCHVOL0026E") + lock = upload_lock_pool[vol_path] = UpdateLock() + return lock.get_lock() + + +def release_vol_update_lock(vol_path): + with pool_lock: + vol_lock = upload_lock_pool.get(vol_path) + if vol_lock.release_lock(): + upload_lock_pool.pop(vol_path, None) + upload_semaphore.release()

On 22/04/2015 05:02, lvroyce@linux.vnet.ibm.com wrote:

From: Royce Lv <lvroyce@linux.vnet.ibm.com>

Storage update abandoned async task because mem copy is not long lasting task and does not need to use task to track its status. So update just do the data upload, and following storage volume lookup will take care of current upload progress.

Signed-off-by: Royce Lv <lvroyce@linux.vnet.ibm.com> --- src/kimchi/i18n.py | 1 + src/kimchi/model/storagevolumes.py | 61 +++++++++++++++++--------------------- 2 files changed, 28 insertions(+), 34 deletions(-)

diff --git a/src/kimchi/i18n.py b/src/kimchi/i18n.py index b712c16..13ac55b 100644 --- a/src/kimchi/i18n.py +++ b/src/kimchi/i18n.py @@ -216,6 +216,7 @@ messages = { "KCHVOL0022E": _("Unable to access file %(url)s. Please, check it."), "KCHVOL0023E": _("Unable to clone storage volume '%(name)s' in pool '%(pool)s'. Details: %(err)s"), "KCHVOL0024E": _("Upload volume chunk index, size and total size must be integer"), + "KCHVOL0025E": _("Upload volume fails. Details: %(err)s"), "KCHVOL0026E": _("Inconsistent upload count"),

"KCHIFACE0001E": _("Interface %(name)s does not exist"), diff --git a/src/kimchi/model/storagevolumes.py b/src/kimchi/model/storagevolumes.py index f02efca..8121324 100644 --- a/src/kimchi/model/storagevolumes.py +++ b/src/kimchi/model/storagevolumes.py @@ -34,6 +34,7 @@ from kimchi.isoinfo import IsoImage from kimchi.model.diskutils import get_disk_ref_cnt from kimchi.model.storagepools import StoragePoolModel from kimchi.model.tasks import TaskModel +from kimchi.model.utils import get_vol_update_lock, release_vol_update_lock from kimchi.utils import add_task, get_next_clone_name, get_unique_file_name from kimchi.utils import kimchi_log from kimchi.xmlutils.utils import xpath_get_text @@ -58,7 +59,7 @@ class StorageVolumesModel(object): self.task = TaskModel(**kargs)

def create(self, pool_name, params): - vol_source = ['file', 'url', 'capacity'] + vol_source = ['url', 'capacity']

It should be joined with the first patch.

name = params.get('name')

@@ -89,9 +90,7 @@ class StorageVolumesModel(object): # if 'name' is omitted - except for the methods listed in # 'REQUIRE_NAME_PARAMS' - the default volume name will be the # file/URL basename. - if create_param == 'file': - name = os.path.basename(params['file'].filename) - elif create_param == 'url': + if create_param == 'url': name = os.path.basename(params['url']) else: name = 'upload-%s' % int(time.time()) @@ -120,36 +119,6 @@ class StorageVolumesModel(object): taskid = add_task(targeturi, create_func, self.objstore, params) return self.task.lookup(taskid)

- def _create_volume_with_file(self, cb, params): - pool_name = params.pop('pool') - dir_path = StoragePoolModel( - conn=self.conn, objstore=self.objstore).lookup(pool_name)['path'] - file_path = os.path.join(dir_path, params['name']) - if os.path.exists(file_path): - raise InvalidParameter('KCHVOL0001E', {'name': params['name']}) - - upload_file = params['file'] - f_len = upload_file.fp.length - try: - size = 0 - with open(file_path, 'wb') as f: - while True: - data = upload_file.file.read(READ_CHUNK_SIZE) - if not data: - break - size += len(data) - f.write(data) - cb('%s/%s' % (size, f_len)) - except Exception as e: - raise OperationFailed('KCHVOL0007E', - {'name': params['name'], - 'pool': pool_name, - 'err': e.message}) - - # Refresh to make sure volume can be found in following lookup - StoragePoolModel.get_storagepool(pool_name, self.conn).refresh(0) - cb('OK', True) - def _create_volume_with_capacity(self, cb, params): pool_name = params.pop('pool') vol_xml = """ @@ -489,6 +458,30 @@ class StorageVolumeModel(object):

cb('OK', True)

+ def update(self, pool, name, params): + vol = StorageVolumeModel.get_storagevolume(pool, name, self.conn) + vol_path = vol.path() + + chunk_size = int(params['chunk_size']) + index = int(params['index']) + pos = chunk_size * index + lock = get_vol_update_lock(vol_path) + try: + with lock: + st = self.conn.get().newStream(0) + vol.upload(st, pos, chunk_size) + st.send(params['chunk'].fullvalue()) + st.finish() + except Exception as e: + st and st.abort() + raise OperationFailed('KCHVOL0025E', + {'err': e.message}) + finally: + release_vol_update_lock(vol_path) + + # Refresh to make sure volume can be found in following lookup + StoragePoolModel.get_storagepool(pool, self.conn).refresh(0) +

class IsoVolumesModel(object): def __init__(self, **kargs):

On 22/04/2015 05:02, lvroyce@linux.vnet.ibm.com wrote:

From: Royce Lv <lvroyce@linux.vnet.ibm.com>

When uploading an ISO file, the ISO distro record may be incomplete because uploading in progress. "call_stack":"Traceback (most recent call last): File \"./src/kimchi/model/storagevolumes.py\", line 304, in lookup iso_img = IsoImage(path) File \"./src/kimchi/isoinfo.py\", line 149, in __init__ self._scan() File \"./src/kimchi/isoinfo.py\", line 437, in _scan self._scan_el_torito(data) File \"./src/kimchi/isoinfo.py\", line 218, in _scan_el_torito ident, csum, key55, keyAA) = self._unpack(fmt, tmp_data) File \"./src/kimchi/isoinfo.py\", line 181, in _unpack return s.unpack(data[:s.size]) error: unpack requires a string argument of length 32\n" So wrap reading record with error handling so that error will not occur in uploading.

Signed-off-by: Royce Lv <lvroyce@linux.vnet.ibm.com> --- src/kimchi/i18n.py | 1 + src/kimchi/isoinfo.py | 5 ++++- 2 files changed, 5 insertions(+), 1 deletion(-)

diff --git a/src/kimchi/i18n.py b/src/kimchi/i18n.py index 13ac55b..c2223f5 100644 --- a/src/kimchi/i18n.py +++ b/src/kimchi/i18n.py @@ -67,6 +67,7 @@ messages = { "to file access control lists for '%(user)s' user if possible, or add the " "'%(user)s' to the ISO path group, or (not recommended) 'chmod -R o+x 'path_to_iso'." "Details: %(err)s" ), + "KCHISO0009E": _("Incomplete record while reading ISO %(filename)s."),

"KCHIMG0001E": _("An error occurred when probing image OS information."), "KCHIMG0002E": _("No OS information found in given image."), diff --git a/src/kimchi/isoinfo.py b/src/kimchi/isoinfo.py index fd1f8f7..f20019d 100644 --- a/src/kimchi/isoinfo.py +++ b/src/kimchi/isoinfo.py @@ -149,7 +149,10 @@ class IsoImage(object): self.remote = self._is_iso_remote() self.volume_id = None self.bootable = False - self._scan() + try: + self._scan() + except: + raise IsoFormatError('KCHISO0009E', {'filename': self.path})

Will this error be displayed to user? If so, I don't think it is needed. I mean, we know it is not completed due a upload process, so we should ignore it.

def _is_iso_remote(self): if os.path.exists(self.path):

On 22/04/2015 05:02, lvroyce@linux.vnet.ibm.com wrote:

From: Royce Lv <lvroyce@linux.vnet.ibm.com>

Signed-off-by: Royce Lv <lvroyce@linux.vnet.ibm.com> --- tests/test_model_storagevolume.py | 95 +++++++++++++++++++++++++++++---------- 1 file changed, 72 insertions(+), 23 deletions(-)

diff --git a/tests/test_model_storagevolume.py b/tests/test_model_storagevolume.py index a3c3ce3..6433a1f 100644 --- a/tests/test_model_storagevolume.py +++ b/tests/test_model_storagevolume.py @@ -20,17 +20,18 @@

import json import os -import requests +import tempfile import unittest

+from cherrypy._cpreqbody import Entity from functools import partial

-from kimchi.config import paths, READONLY_POOL_TYPE +from kimchi.config import READONLY_POOL_TYPE from kimchi.model.model import Model from kimchi.mockmodel import MockModel from kimchi.rollbackcontext import RollbackContext -from utils import fake_auth_header, get_free_port, patch_auth, request -from utils import rollback_wrapper, run_server, wait_task +from utils import get_free_port, patch_auth, request +from utils import rollback_wrapper, run_server, wait_task, running_as_root

model = None @@ -149,25 +150,6 @@ def _do_volume_test(self, model, host, ssl_port, pool_name): resp = self.request(vol_uri) self.assertEquals(404, resp.status)

- # Create storage volume with 'file' - filepath = os.path.join(paths.get_prefix(), 'COPYING.LGPL') - url = 'https://%s:%s' % (host, ssl_port) + uri - with open(filepath, 'rb') as fd: - r = requests.post(url, files={'file': fd}, - verify=False, - headers=fake_auth_header()) - - if pool_info['type'] in READONLY_POOL_TYPE: - self.assertEquals(r.status_code, 400) - else: - rollback.prependDefer(model.storagevolume_delete, pool_name, - 'COPYING.LGPL') - self.assertEquals(r.status_code, 202) - task = r.json() - wait_task(_task_lookup, task['id']) - resp = self.request(uri + '/COPYING.LGPL') - self.assertEquals(200, resp.status) -

This should be added to the first patch of this series.

# Create storage volume with 'url' url = 'https://github.com/kimchi-project/kimchi/raw/master/COPYING' req = json.dumps({'url': url}) @@ -189,6 +171,73 @@ class StorageVolumeTests(unittest.TestCase): def setUp(self): self.request = partial(request, host, ssl_port)

+ @unittest.skipUnless(running_as_root(), 'Must be run as root') + def test_volume_upload(self):

+ class fake_header(object): + def elements(self, key): + return None + + def __getattr__(self, key): + try: + return self.__getattribute__(key) + except AttributeError: + return lambda x, y: None +

For what is this for?

+ #inst = Model(objstore_loc=self.tmp_store)

You can remove this comment.

+ vol_path = os.path.abspath('./run_tests.sh') + + def do_upload(tmp_file, vol_path, url): + index = 0 + chunk_size = 2 * 1000 + with open(vol_path, 'rb') as fd: + while True: + with open(tmp_file, 'wb') as tmp_fd: + fd.seek(index * chunk_size) + data = fd.read(chunk_size) + tmp_fd.write(data) + + # only file open for read can be handled by cherrypy + with open(tmp_file, 'rb') as tmp_fd: + # Hack cherrypy entity object and pass it to volume upload + entity = Entity(None, fake_header()) + entity.file = tmp_fd + param = {'index': str(index), + 'chunk_size': str(chunk_size), + 'chunk': entity}

+ model.storagevolume_update(pool, params['name'], param) + vol_info = model.storagevolume_lookup(pool, params['name'])

We have agreed to always do the tests using the REST API instead of the model instances. So please, change it to request the data using the REST API

+ index = index + 1 + if len(data) < chunk_size: + return vol_info + + # Create a volume with raw format first, following upload will override it. + params = {'capacity': os.path.getsize(vol_path), + 'format': 'raw', + 'name': os.path.basename(vol_path)} + pool = 'default' + + with RollbackContext() as rollback:

+ task_response = model.storagevolumes_create(pool, params) + rollback.prependDefer(model.storagevolume_delete, pool, + params['name']) + taskid = task_response['id'] + vol_uri = task_response['target_uri'] + model.task_wait(taskid) + self.assertEquals('finished', model.task_lookup(taskid)['status']) +

Same I commented above related to use the REST API instead of the model instance.

+ f = tempfile.NamedTemporaryFile(delete=False) + rollback.prependDefer(os.remove, f.name) + resp = do_upload(f.name, vol_path, vol_uri) + + with open(vol_path) as vol_file: + vol_content = vol_file.read() + + with open(resp['path']) as copy_file: + cp_content = copy_file.read() + + self.assertEquals(vol_content, cp_content) + + def test_get_storagevolume(self): uri = '/storagepools/default/storagevolumes' resp = self.request(uri)

On 22/04/2015 05:02, lvroyce@linux.vnet.ibm.com wrote:

From: Royce Lv <lvroyce@linux.vnet.ibm.com>

This is a test script against real model server, you can use it to verify the logic of upload. NOTE: pls change the username/password/file path to a valid one.

Signed-off-by: Royce Lv <lvroyce@linux.vnet.ibm.com> --- tests/test_api.py | 108 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) create mode 100644 tests/test_api.py

diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..953716a --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,108 @@ +# -*- coding: utf-8 -*- +# +# Project Kimchi +# +# Copyright IBM, Corp. 2013-2014 +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library 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 +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + +import base64 +import json +import os +import requests +import unittest +import time + +from functools import partial + +from utils import request + + +host = None +ssl_port = None + + +def setUpModule(): + global host, ssl_port + + host = '127.0.0.1' + ssl_port = 8001 + + +class RestTests(unittest.TestCase): + def setUp(self): + self.request = partial(request, host, ssl_port) + + def assertHTTPStatus(self, code, *args): + resp = self.request(*args) + self.assertEquals(code, resp.status) + + def assertValidJSON(self, txt): + try: + json.loads(txt) + except ValueError: + self.fail("Invalid JSON: %s" % txt) +

+ def test_upload(self): + # If we use self.request, we may encode multipart formdata by ourselves + # requests lib take care of encode part, so use this lib instead + def fake_auth_header(): + headers = {'Accept': 'application/json'} + user, pw = ('this is your user', 'this is your password') + hdr = "Basic " + base64.b64encode("%s:%s" % (user, pw)) + headers['AUTHORIZATION'] = hdr + return headers + + def do_upload(vol_path, url): + index = 0 + chunk_size = 2 * 1000 * 1000 + with open(vol_path, 'rb') as fd: + while True: + with open(vol_path + '.tmp', 'wb') as tmp_fd: + # make sure it is truncated + fd.seek(index*chunk_size) + data = fd.read(chunk_size) + tmp_fd.write(data) + + with open(vol_path + '.tmp', 'rb') as tmp_fd: + print "size of data %s" % len(data) + r = requests.put(url, + data={"index": str(index), "chunk_size": str(chunk_size)}, + files={"chunk": tmp_fd}, + verify=False, + headers=fake_auth_header()) + self.assertEquals(r.status_code, 200) + index = index + 1 + print "index is %s" % index + if len(data) < chunk_size: + return index + + def do_thousands(url): + for i in range(1000): + self.request(url, headers=fake_auth_header()) + + vol_path = os.path.join('/home/royce/upload/vm2.qcow2') + file_len = os.path.getsize(vol_path) + url = "https://%s:%s/storagepools/default/storagevolumes/" % \ + (host, ssl_port) + body = json.dumps({'capacity': file_len, 'name': os.path.basename(vol_path), 'format': 'raw'}) + r = requests.post(url, + data=body, + verify=False, + headers=fake_auth_header()) + time.sleep(1) + print r.text + url = url + os.path.basename(vol_path) + do_upload(vol_path, url)

This test should be added to the Kimchi test cases suite.