Merge pull request #121 from Storj/develop

Hotfix missing parametr storageSize

Author: steenzout

storj/model.py

@@ -181,11 +181,12 @@ class Frame(Object):
         shards (list[:py:class:`Shard`]): shards that compose this frame.
     """
 
-    def __init__(self, id=None, created=None, shards=None, locked=None, user=None, size=None):
+    def __init__(self, id=None, created=None, shards=None, locked=None, user=None, size=None, storageSize=None):
         self.id = id
         self.locked = locked
         self.user = user
         self.size = size
+        self.storageSize = storageSize
 
         if created is not None:
             self.created = datetime.fromtimestamp(

storj/uploader.py

@@ -18,10 +18,39 @@
 from exception import BridgeError, FarmerError, SuppliedTokenNotAcceptedError
 from http import Client
 
+import threading
+import thread
+
+TIMEOUT = 60    # default = 1 minute
+
 
 def foo(args):
     self, shard, shard_index, frame, file_name, tmp_path = args
-    self.upload_shard(shard, shard_index, frame, file_name, tmp_path)
+    return self.upload_shard(shard, shard_index, frame, file_name, tmp_path)
+
+
+def quit_function(fn_name):
+    # self.__logger.debug('{0} took too long'.format(fn_name))
+    thread.interrupt_main()  # raises KeyboardInterrupt
+
+
+def exit_after(s):
+    '''
+    use as decorator to exit process if
+    function takes longer than s seconds
+    '''
+    def outer(fn):
+        def inner(*args, **kwargs):
+            timer = threading.Timer(s, quit_function, args=[fn.__name__])
+            timer.start()
+            try:
+                result = fn(*args, **kwargs)
+            finally:
+                timer.cancel()
+            return result
+        return inner
+
+    return outer
 
 
 class Uploader:
@@ -101,6 +130,26 @@ def _prepare_bucket_entry_hmac(self, shard_array):
 
         return current_hmac
 
+    @exit_after(TIMEOUT)
+    def require_upload(self, shard_path, url, index):
+        with open(shard_path, 'rb') as f:
+            response = requests.post(
+                url,
+                data=self._read_in_chunks(
+                    f, shard_index=index),
+                timeout=1)
+            return response
+
+    def _calculate_timeout(self, shard_size, mbps=0.5):
+        """
+        Args:
+            shard_size: shard size in Byte
+            mbps: upload throughtput. Default 500 kbps
+        """
+        global TIMEOUT
+        TIMEOUT = int(shard_size * 8.0 / (1024 ** 2 * mbps))
+        self.__logger.debug('Set timeout to %s seconds' % TIMEOUT)
+
     def upload_shard(self, shard, chapters, frame,
                      file_name_ready_to_shard_upload, tmp_path):
         """
@@ -120,7 +169,7 @@ def upload_shard(self, shard, chapters, frame,
             contract_negotiation_tries += 1
             self.__logger.debug('Negotiating contract')
             self.__logger.debug('Trying to negotiate storage contract for \
-shard at index %s...' % chapters)
+shard at index %s. Attempt %s' % (chapters, contract_negotiation_tries))
 
             try:
                 frame_content = self.client.frame_add_shard(shard, frame.id)
@@ -132,7 +181,9 @@ def upload_shard(self, shard, chapters, frame,
                     frame_content['farmer']['port'],
                     frame_content['hash'],
                     frame_content['token'])
-                self.__logger.debug('upload_shard url=%s', url)
+                self.__logger.debug('Done contract for shard %s with url=%s',
+                                    chapters,
+                                    url)
 
                 # begin recording exchange report
                 # exchange_report = model.ExchangeReport()
@@ -151,66 +202,82 @@ def upload_shard(self, shard, chapters, frame,
 
                     try:
                         self.__logger.debug(
-                            'Upload shard at index %s to %s:%d attempt #%d',
+                            'Upload shard at index %s to %s attempt #%d',
                             shard.index,
                             frame_content['farmer']['address'],
-                            frame_content['farmer']['port'],
                             farmer_tries)
 
                         mypath = os.path.join(
                             tmp_path, '%s-%s' % (
                                 file_name_ready_to_shard_upload,
                                 chapters + 1))
 
+                        """
                         with open(mypath, 'rb') as f:
                             response = requests.post(
                                 url,
                                 data=self._read_in_chunks(
                                     f, shard_index=chapters),
                                 timeout=1)
+                        """
+                        response = self.require_upload(mypath, url, chapters)
+                        self.__logger.debug('>>> Shard %s Uploaded' % chapters)
 
                         j = json.loads(str(response.content))
+                        self.__logger.info('>>>> %s' % str(j))
 
                         if j.get('result') == \
                                 'The supplied token is not accepted':
                             raise SuppliedTokenNotAcceptedError()
 
+                    # Exceptions raised when uploading shards
                     except FarmerError as e:
+                        self.__logger.error('Farmer error')
                         self.__logger.error(e)
                         continue
 
+                    except KeyboardInterrupt:
+                        self.__logger.error()
+                        self.__logger.error()
+                        self.__logger.error(
+                            'Upload shard %s to %s too slow.' % (
+                                chapters, url))
+                        self.__logger.error(
+                            'Upload timed out. Redo upload of shard %s' %
+                            chapters)
+                        continue
+
                     except Exception as e:
+                        self.__logger.error('Exception')
                         self.__logger.error(e)
                         self.__logger.error(
-                            'Shard upload error for to %s:%d',
+                            'Shard upload error for %s to %s:%d',
+                            chapters,
                             frame_content['farmer']['address'],
                             frame_content['farmer']['port'])
                         continue
 
-                    self.shards_already_uploaded += 1
-                    self.__logger.info(
-                        'Shard uploaded successfully to %s:%d',
-                        frame_content['farmer']['address'],
-                        frame_content['farmer']['port'])
-
-                    self.__logger.debug(
-                        '%s shards, %s sent',
-                        self.all_shards_count,
-                        self.shards_already_uploaded)
-
-                    if int(self.all_shards_count) <= \
-                            int(self.shards_already_uploaded):
-                        self.__logger.debug('finish upload')
+                    else:
+                        self.shards_already_uploaded += 1
+                        self.__logger.info(
+                            'Shard uploaded successfully to %s:%d',
+                            frame_content['farmer']['address'],
+                            frame_content['farmer']['port'])
 
-                    break
+                        self.__logger.debug(
+                            '%s shards, %s sent',
+                            self.all_shards_count,
+                            self.shards_already_uploaded)
 
-                self.__logger.debug('response.content=%s', response.content)
+                        if int(self.all_shards_count) <= \
+                                int(self.shards_already_uploaded):
+                            self.__logger.debug('finish upload')
 
-                j = json.loads(str(response.content))
-                if j.get('result') == 'The supplied token is not accepted':
-                    raise SuppliedTokenNotAcceptedError()
+                        break
 
+            # Exceptions raised negotiating contracts
             except BridgeError as e:
+                self.__logger.error('Bridge error')
                 self.__logger.error(e)
 
                 # upload failed due to Storj Bridge failure
@@ -236,20 +303,22 @@ def upload_shard(self, shard, chapters, frame,
                 # Send exchange report
                 # self.client.send_exchange_report(exchange_report)
                 continue
+            else:
+                # uploaded with success
+                current_timestamp = int(time.time())
+                # prepare second half of exchange heport
+                exchange_report.exchangeEnd = str(current_timestamp)
+                exchange_report.exchangeResultCode = exchange_report.SUCCESS
+                exchange_report.exchangeResultMessage = \
+                    exchange_report.STORJ_REPORT_SHARD_UPLOADED
 
-            # uploaded with success
-            current_timestamp = int(time.time())
-            # prepare second half of exchange heport
-            exchange_report.exchangeEnd = str(current_timestamp)
-            exchange_report.exchangeResultCode = exchange_report.SUCCESS
-            exchange_report.exchangeResultMessage = \
-                exchange_report.STORJ_REPORT_SHARD_UPLOADED
-
-            self.__logger.info('Shard %s successfully added and exchange \
-report sent.', chapters + 1)
-            # Send exchange report
-            # self.client.send_exchange_report(exchange_report)
-            break
+                self.__logger.info('Shard %s successfully added and exchange \
+report sent.    ', chapters + 1)
+                # Send exchange report
+                # self.client.send_exchange_report(exchange_report)
+                # break
+                return True
+            return False
 
     def _read_in_chunks(self, file_object, blocksize=4096, chunks=-1,
                         shard_index=None):
@@ -277,11 +346,9 @@ def _read_in_chunks(self, file_object, blocksize=4096, chunks=-1,
     def file_upload(self, bucket_id, file_path, tmp_file_path):
         """"""
 
-        self.__logger.debug('Upload %s in bucket %d', file_path, bucket_id)
+        self.__logger.debug('Upload %s in bucket %s', file_path, bucket_id)
         self.__logger.debug('Temp folder %s', tmp_file_path)
 
-        encryption_enabled = True
-
         bname = os.path.split(file_path)[1]  # File name
 
         file_mime_type = 'text/plain'
@@ -343,18 +410,29 @@ def file_upload(self, bucket_id, file_path, tmp_file_path):
 
         self.__logger.debug('Sharding ended...')
 
-        self.__logger.debug('There are %d shards', self.all_shards_count)
+        self.__logger.debug('There are %s shards', self.all_shards_count)
 
-        # create file hash
+        # Calculate timeout
+        self._calculate_timeout(shard_size=shards_manager.shards[0].size,
+                                mbps=1)
 
+        # Upload shards
         mp = Pool()
-        mp.map(foo, [(self, shards_manager.shards[x], x, frame,
-                      file_name_ready_to_shard_upload, tmp_file_path)
-                     for x in range(len(shards_manager.shards))])
-
+        res = mp.map(foo, [(self, shards_manager.shards[x], x, frame,
+                           file_name_ready_to_shard_upload, tmp_file_path)
+                           for x in range(len(shards_manager.shards))])
+
+        self.__logger.debug('===== RESULTS =====')
+        self.__logger.debug(res)
+        if False in res:
+            self.__logger.error('File not uploaded: shard %s not uploaded' %
+                                res.index(False))
+            self.__logger.error('Exiting with errors')
+            exit(1)
         # finish_upload
         self.__logger.debug('Generating HMAC...')
 
+        # create file hash
         hash_sha512_hmac_b64 = self._prepare_bucket_entry_hmac(
             shards_manager.shards)
         hash_sha512_hmac = hashlib.sha224(str(