Synnefo » ./kamaki

# Copyright 2011-2013 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

from threading import enumerate as activethreads

from os import fstat

from hashlib import new as newhashlib

from time import time

from StringIO import StringIO

from binascii import hexlify

from kamaki.clients import SilentEvent, sendlog

from kamaki.clients.pithos.rest_api import PithosRestClient

from kamaki.clients.storage import ClientError

from kamaki.clients.utils import path4url, filter_in

def _pithos_hash(block, blockhash):

    h = newhashlib(blockhash)

    h.update(block.rstrip('\x00'))

    return h.hexdigest()

def _range_up(start, end, max_value, a_range):

    """

    :param start: (int) the window bottom

    :param end: (int) the window top

    :param max_value: (int) maximum accepted value

    :param a_range: (str) a range string in the form X[,X'[,X''[...]]]

        where X: x|x-y|-x where x < y and x, y natural numbers

    :returns: (str) a range string cut-off for the start-end range

        an empty response means this window is out of range

    """

    assert start >= 0, '_range_up was called with start < 0'

    assert end >= start, '_range_up was called with end < start'

    assert end <= max_value, '_range_up was called with max_value < end'

    if not a_range:

        return '%s-%s' % (start, end)

    selected = []

    for some_range in a_range.split(','):

        v0, sep, v1 = some_range.partition('-')

        if v0:

            v0 = int(v0)

            if sep:

                v1 = int(v1)

                if v1 < start or v0 > end or v1 < v0:

                    continue

                v0 = v0 if v0 > start else start

                v1 = v1 if v1 < end else end

                selected.append('%s-%s' % (v0, v1))

            elif v0 < start:

                continue

            else:

                v1 = v0 if v0 <= end else end

                selected.append('%s-%s' % (start, v1))

        else:

            v1 = int(v1)

            if max_value - v1 > end:

                continue

            v0 = (max_value - v1) if max_value - v1 > start else start

            selected.append('%s-%s' % (v0, end))

    return ','.join(selected)

class PithosClient(PithosRestClient):

    """Synnefo Pithos+ API client"""

    def __init__(self, base_url, token, account=None, container=None):

        super(PithosClient, self).__init__(base_url, token, account, container)

    def create_container(

            self,

            container=None, sizelimit=None, versioning=None, metadata=None):

        """

        :param container: (str) if not given, self.container is used instead

        :param sizelimit: (int) container total size limit in bytes

        :param versioning: (str) can be auto or whatever supported by server

        :param metadata: (dict) Custom user-defined metadata of the form

            { 'name1': 'value1', 'name2': 'value2', ... }

        :returns: (dict) response headers

        """

        cnt_back_up = self.container

        try:

            self.container = container or cnt_back_up

            r = self.container_put(

                quota=sizelimit, versioning=versioning, metadata=metadata)

            return r.headers

        finally:

            self.container = cnt_back_up

    def purge_container(self, container=None):

        """Delete an empty container and destroy associated blocks

        """

        cnt_back_up = self.container

        try:

            self.container = container or cnt_back_up

            r = self.container_delete(until=unicode(time()))

        finally:

            self.container = cnt_back_up

        return r.headers

    def upload_object_unchunked(

            self, obj, f,

            withHashFile=False,

            size=None,

            etag=None,

            content_encoding=None,

            content_disposition=None,

            content_type=None,

            sharing=None,

            public=None):

        """

        :param obj: (str) remote object path

        :param f: open file descriptor

        :param withHashFile: (bool)

        :param size: (int) size of data to upload

        :param etag: (str)

        :param content_encoding: (str)

        :param content_disposition: (str)

        :param content_type: (str)

        :param sharing: {'read':[user and/or grp names],

            'write':[usr and/or grp names]}

        :param public: (bool)

        :returns: (dict) created object metadata

        """

        self._assert_container()

        if withHashFile:

            data = f.read()

            try:

                import json

                data = json.dumps(json.loads(data))

            except ValueError:

                raise ClientError('"%s" is not json-formated' % f.name, 1)

            except SyntaxError:

                msg = '"%s" is not a valid hashmap file' % f.name

                raise ClientError(msg, 1)

            f = StringIO(data)

        else:

            data = f.read(size) if size else f.read()

        r = self.object_put(

            obj,

            data=data,

            etag=etag,

            content_encoding=content_encoding,

            content_disposition=content_disposition,

            content_type=content_type,

            permissions=sharing,

            public=public,

            success=201)

        return r.headers

    def create_object_by_manifestation(

            self, obj,

            etag=None,

            content_encoding=None,

            content_disposition=None,

            content_type=None,

            sharing=None,

            public=None):

        """

        :param obj: (str) remote object path

        :param etag: (str)

        :param content_encoding: (str)

        :param content_disposition: (str)

        :param content_type: (str)

        :param sharing: {'read':[user and/or grp names],

            'write':[usr and/or grp names]}

        :param public: (bool)

        :returns: (dict) created object metadata

        """

        self._assert_container()

        r = self.object_put(

            obj,

            content_length=0,

            etag=etag,

            content_encoding=content_encoding,

            content_disposition=content_disposition,

            content_type=content_type,

            permissions=sharing,

            public=public,

            manifest='%s/%s' % (self.container, obj))

        return r.headers

    # upload_* auxiliary methods

    def _put_block_async(self, data, hash):

        event = SilentEvent(method=self._put_block, data=data, hash=hash)

        event.start()

        return event

    def _put_block(self, data, hash):

        r = self.container_post(

            update=True,

            content_type='application/octet-stream',

            content_length=len(data),

            data=data,

            format='json')

        assert r.json[0] == hash, 'Local hash does not match server'

    def _get_file_block_info(self, fileobj, size=None, cache=None):

        """

        :param fileobj: (file descriptor) source

        :param size: (int) size of data to upload from source

        :param cache: (dict) if provided, cache container info response to

        avoid redundant calls

        """

        if isinstance(cache, dict):

            try:

                meta = cache[self.container]

            except KeyError:

                meta = self.get_container_info()

                cache[self.container] = meta

        else:

            meta = self.get_container_info()

        blocksize = int(meta['x-container-block-size'])

        blockhash = meta['x-container-block-hash']

        size = size if size is not None else fstat(fileobj.fileno()).st_size

        nblocks = 1 + (size - 1) // blocksize

        return (blocksize, blockhash, size, nblocks)

    def _create_object_or_get_missing_hashes(

            self, obj, json,

            size=None,

            format='json',

            hashmap=True,

            content_type=None,

            if_etag_match=None,

            if_etag_not_match=None,

            content_encoding=None,

            content_disposition=None,

            permissions=None,

            public=None,

            success=(201, 409)):

        r = self.object_put(

            obj,

            format='json',

            hashmap=True,

            content_type=content_type,

            json=json,

            if_etag_match=if_etag_match,

            if_etag_not_match=if_etag_not_match,

            content_encoding=content_encoding,

            content_disposition=content_disposition,

            permissions=permissions,

            public=public,

            success=success)

        return (None if r.status_code == 201 else r.json), r.headers

    def _calculate_blocks_for_upload(

            self, blocksize, blockhash, size, nblocks, hashes, hmap, fileobj,

            hash_cb=None):

        offset = 0

        if hash_cb:

            hash_gen = hash_cb(nblocks)

            hash_gen.next()

        for i in range(nblocks):

            block = fileobj.read(min(blocksize, size - offset))

            bytes = len(block)

            hash = _pithos_hash(block, blockhash)

            hashes.append(hash)

            hmap[hash] = (offset, bytes)

            offset += bytes

            if hash_cb:

                hash_gen.next()

        msg = 'Failed to calculate uploaded blocks:'

        ' Offset and object size do not match'

        assert offset == size, msg

    def _upload_missing_blocks(self, missing, hmap, fileobj, upload_gen=None):

        """upload missing blocks asynchronously"""

        self._init_thread_limit()

        flying = []

        failures = []

        for hash in missing:

            offset, bytes = hmap[hash]

            fileobj.seek(offset)

            data = fileobj.read(bytes)

            r = self._put_block_async(data, hash)

            flying.append(r)

            unfinished = self._watch_thread_limit(flying)

            for thread in set(flying).difference(unfinished):

                if thread.exception:

                    failures.append(thread)

                    if isinstance(

                            thread.exception,

                            ClientError) and thread.exception.status == 502:

                        self.POOLSIZE = self._thread_limit

                elif thread.isAlive():

                    flying.append(thread)

                elif upload_gen:

                    try:

                        upload_gen.next()

                    except:

                        pass

            flying = unfinished

        for thread in flying:

            thread.join()

            if thread.exception:

                failures.append(thread)

            elif upload_gen:

                try:

                    upload_gen.next()

                except:

                    pass

        return [failure.kwargs['hash'] for failure in failures]

    def upload_object(

            self, obj, f,

            size=None,

            hash_cb=None,

            upload_cb=None,

            etag=None,

            if_etag_match=None,

            if_not_exist=None,

            content_encoding=None,

            content_disposition=None,

            content_type=None,

            sharing=None,

            public=None,

            container_info_cache=None):

        """Upload an object using multiple connections (threads)

        :param obj: (str) remote object path

        :param f: open file descriptor (rb)

        :param hash_cb: optional progress.bar object for calculating hashes

        :param upload_cb: optional progress.bar object for uploading

        :param etag: (str)

        :param if_etag_match: (str) Push that value to if-match header at file

            creation

        :param if_not_exist: (bool) If true, the file will be uploaded ONLY if

            it does not exist remotely, otherwise the operation will fail.

            Involves the case of an object with the same path is created while

            the object is being uploaded.

        :param content_encoding: (str)

        :param content_disposition: (str)

        :param content_type: (str)

        :param sharing: {'read':[user and/or grp names],

            'write':[usr and/or grp names]}

        :param public: (bool)

        :param container_info_cache: (dict) if given, avoid redundant calls to

            server for container info (block size and hash information)

        """

        self._assert_container()

        block_info = (

            blocksize, blockhash, size, nblocks) = self._get_file_block_info(

                f, size, container_info_cache)

        (hashes, hmap, offset) = ([], {}, 0)

        if not content_type:

            content_type = 'application/octet-stream'

        self._calculate_blocks_for_upload(

            *block_info,

            hashes=hashes,

            hmap=hmap,

            fileobj=f,

            hash_cb=hash_cb)

        hashmap = dict(bytes=size, hashes=hashes)

        missing, obj_headers = self._create_object_or_get_missing_hashes(

            obj, hashmap,

            content_type=content_type,

            size=size,

            if_etag_match=if_etag_match,

            if_etag_not_match='*' if if_not_exist else None,

            content_encoding=content_encoding,

            content_disposition=content_disposition,

            permissions=sharing,

            public=public)

        if missing is None:

            return obj_headers

        if upload_cb:

            upload_gen = upload_cb(len(missing))

            for i in range(len(missing), len(hashmap['hashes']) + 1):

                try:

                    upload_gen.next()

                except:

                    upload_gen = None

        else:

            upload_gen = None

        retries = 7

        try:

            while retries:

                sendlog.info('%s blocks missing' % len(missing))

                num_of_blocks = len(missing)

                missing = self._upload_missing_blocks(

                    missing,

                    hmap,

                    f,

                    upload_gen)

                if missing:

                    if num_of_blocks == len(missing):

                        retries -= 1

                    else:

                        num_of_blocks = len(missing)

                else:

                    break

            if missing:

                try:

                    details = ['%s' % thread.exception for thread in missing]

                except Exception:

                    details = ['Also, failed to read thread exceptions']

                raise ClientError(

                    '%s blocks failed to upload' % len(missing),

                    details=details)

        except KeyboardInterrupt:

            sendlog.info('- - - wait for threads to finish')

            for thread in activethreads():

                thread.join()

            raise

        r = self.object_put(

            obj,

            format='json',

            hashmap=True,

            content_type=content_type,

            content_encoding=content_encoding,

            if_etag_match=if_etag_match,

            if_etag_not_match='*' if if_not_exist else None,

            etag=etag,

            json=hashmap,

            permissions=sharing,

            public=public,

            success=201)

        return r.headers

    def upload_from_string(

            self, obj, input_str,

            hash_cb=None,

            upload_cb=None,

            etag=None,

            if_etag_match=None,

            if_not_exist=None,

            content_encoding=None,

            content_disposition=None,

            content_type=None,

            sharing=None,

            public=None,

            container_info_cache=None):

        """Upload an object using multiple connections (threads)

        :param obj: (str) remote object path

        :param input_str: (str) upload content

        :param hash_cb: optional progress.bar object for calculating hashes

        :param upload_cb: optional progress.bar object for uploading

        :param etag: (str)

        :param if_etag_match: (str) Push that value to if-match header at file

            creation

        :param if_not_exist: (bool) If true, the file will be uploaded ONLY if

            it does not exist remotely, otherwise the operation will fail.

            Involves the case of an object with the same path is created while

            the object is being uploaded.

        :param content_encoding: (str)

        :param content_disposition: (str)

        :param content_type: (str)

        :param sharing: {'read':[user and/or grp names],

            'write':[usr and/or grp names]}

        :param public: (bool)

        :param container_info_cache: (dict) if given, avoid redundant calls to

            server for container info (block size and hash information)

        """

        self._assert_container()

        blocksize, blockhash, size, nblocks = self._get_file_block_info(

                fileobj=None, size=len(input_str), cache=container_info_cache)

        (hashes, hmap, offset) = ([], {}, 0)

        if not content_type:

            content_type = 'application/octet-stream'

        hashes = []

        hmap = {}

        for blockid in range(nblocks):

            start = blockid * blocksize

            block = input_str[start: (start + blocksize)]

            hashes.append(_pithos_hash(block, blockhash))

            hmap[hashes[blockid]] = (start, block)

        hashmap = dict(bytes=size, hashes=hashes)

        missing, obj_headers = self._create_object_or_get_missing_hashes(

            obj, hashmap,

            content_type=content_type,

            size=size,

            if_etag_match=if_etag_match,

            if_etag_not_match='*' if if_not_exist else None,

            content_encoding=content_encoding,

            content_disposition=content_disposition,

            permissions=sharing,

            public=public)

        if missing is None:

            return obj_headers

        num_of_missing = len(missing)

        if upload_cb:

            self.progress_bar_gen = upload_cb(nblocks)

            for i in range(nblocks + 1 - num_of_missing):

                self._cb_next()

        tries = 7

        old_failures = 0

        try:

            while tries and missing:

                flying = []

                failures = []

                for hash in missing:

                    offset, block = hmap[hash]

                    bird = self._put_block_async(block, hash)

                    flying.append(bird)

                    unfinished = self._watch_thread_limit(flying)

                    for thread in set(flying).difference(unfinished):

                        if thread.exception:

                            failures.append(thread.kwargs['hash'])

                        if thread.isAlive():

                            flying.append(thread)

                        else:

                            self._cb_next()

                    flying = unfinished

                for thread in flying:

                    thread.join()

                    if thread.exception:

                        failures.append(thread.kwargs['hash'])

                    self._cb_next()

                missing = failures

                if missing and len(missing) == old_failures:

                    tries -= 1

                old_failures = len(missing)

            if missing:

                raise ClientError(

                    '%s blocks failed to upload' % len(missing),

                    details=['%s' % thread.exception for thread in missing])

        except KeyboardInterrupt:

            sendlog.info('- - - wait for threads to finish')

            for thread in activethreads():

                thread.join()

            raise

        r = self.object_put(

            obj,

            format='json',

            hashmap=True,

            content_type=content_type,

            content_encoding=content_encoding,

            if_etag_match=if_etag_match,

            if_etag_not_match='*' if if_not_exist else None,

            etag=etag,

            json=hashmap,

            permissions=sharing,

            public=public,

            success=201)

        return r.headers

    # download_* auxiliary methods

    def _get_remote_blocks_info(self, obj, **restargs):

        #retrieve object hashmap

        myrange = restargs.pop('data_range', None)

        hashmap = self.get_object_hashmap(obj, **restargs)

        restargs['data_range'] = myrange

        blocksize = int(hashmap['block_size'])

        blockhash = hashmap['block_hash']

        total_size = hashmap['bytes']

        #assert total_size/blocksize + 1 == len(hashmap['hashes'])

        map_dict = {}

        for i, h in enumerate(hashmap['hashes']):

            #  map_dict[h] = i   CHAGE

            if h in map_dict:

                map_dict[h].append(i)

            else:

                map_dict[h] = [i]

        return (blocksize, blockhash, total_size, hashmap['hashes'], map_dict)

    def _dump_blocks_sync(

            self, obj, remote_hashes, blocksize, total_size, dst, crange,

            **args):

        for blockid, blockhash in enumerate(remote_hashes):

            if blockhash:

                start = blocksize * blockid

                is_last = start + blocksize > total_size

                end = (total_size - 1) if is_last else (start + blocksize - 1)

                data_range = _range_up(start, end, total_size, crange)

                if not data_range:

                    self._cb_next()

                    continue

                args['data_range'] = 'bytes=%s' % data_range

                r = self.object_get(obj, success=(200, 206), **args)

                self._cb_next()

                dst.write(r.content)

                dst.flush()

    def _get_block_async(self, obj, **args):

        event = SilentEvent(self.object_get, obj, success=(200, 206), **args)

        event.start()

        return event

    def _hash_from_file(self, fp, start, size, blockhash):

        fp.seek(start)

        block = fp.read(size)

        h = newhashlib(blockhash)

        h.update(block.strip('\x00'))

        return hexlify(h.digest())

    def _thread2file(self, flying, blockids, local_file, offset=0, **restargs):

        """write the results of a greenleted rest call to a file

        :param offset: the offset of the file up to blocksize

        - e.g. if the range is 10-100, all blocks will be written to

        normal_position - 10

        """

        for key, g in flying.items():

            if g.isAlive():

                continue

            if g.exception:

                raise g.exception

            block = g.value.content

            for block_start in blockids[key]:

                local_file.seek(block_start + offset)

                local_file.write(block)

                self._cb_next()

            flying.pop(key)

            blockids.pop(key)

        local_file.flush()

    def _dump_blocks_async(

            self, obj, remote_hashes, blocksize, total_size, local_file,

            blockhash=None, resume=False, filerange=None, **restargs):

        file_size = fstat(local_file.fileno()).st_size if resume else 0

        flying = dict()

        blockid_dict = dict()

        offset = 0

        self._init_thread_limit()

        for block_hash, blockids in remote_hashes.items():

            blockids = [blk * blocksize for blk in blockids]

            unsaved = [blk for blk in blockids if not (

                blk < file_size and block_hash == self._hash_from_file(

                        local_file, blk, blocksize, blockhash))]

            self._cb_next(len(blockids) - len(unsaved))

            if unsaved:

                key = unsaved[0]

                self._watch_thread_limit(flying.values())

                self._thread2file(

                    flying, blockid_dict, local_file, offset,

                    **restargs)

                end = total_size - 1 if (

                    key + blocksize > total_size) else key + blocksize - 1

                data_range = _range_up(key, end, total_size, filerange)

                if not data_range:

                    self._cb_next()

                    continue

                restargs['async_headers'] = {'Range': 'bytes=%s' % data_range}

                flying[key] = self._get_block_async(obj, **restargs)

                blockid_dict[key] = unsaved

        for thread in flying.values():

            thread.join()

        self._thread2file(flying, blockid_dict, local_file, offset, **restargs)

    def download_object(

            self, obj, dst,

            download_cb=None,

            version=None,

            resume=False,

            range_str=None,

            if_match=None,

            if_none_match=None,

            if_modified_since=None,

            if_unmodified_since=None):

        """Download an object (multiple connections, random blocks)

        :param obj: (str) remote object path

        :param dst: open file descriptor (wb+)

        :param download_cb: optional progress.bar object for downloading

        :param version: (str) file version

        :param resume: (bool) if set, preserve already downloaded file parts

        :param range_str: (str) from, to are file positions (int) in bytes

        :param if_match: (str)

        :param if_none_match: (str)

        :param if_modified_since: (str) formated date

        :param if_unmodified_since: (str) formated date"""

        restargs = dict(

            version=version,

            data_range=None if range_str is None else 'bytes=%s' % range_str,

            if_match=if_match,

            if_none_match=if_none_match,

            if_modified_since=if_modified_since,

            if_unmodified_since=if_unmodified_since)

        (

            blocksize,

            blockhash,

            total_size,

            hash_list,

            remote_hashes) = self._get_remote_blocks_info(obj, **restargs)