Ignore "[Errno 29] Illegal seek" errors which may arise when using "upload_object_via_stream" method

CHANGES.rst

@@ -40,6 +40,17 @@ Storage
   (GITHUB-1417, GITHUB-1418)
   [Tomaz Muraus]
 
+- Fix ``upload_object_via_stream`` method so "Illegal seek" errors which
+  can arise when calculating iterator content hash are ignored. Those errors
+  likely indicate that the underlying file handle / iterator is a pipe which
+  doesn't support seek and that the error is not fatal and we should still
+  proceed.
+
+  Reported by Per Buer - @perbu.
+
+  (GITHUB-1424, GITHUB-1427)
+  [Tomaz Muraus]
+
 DNS
 ~~~
 
@@ -49,6 +60,7 @@ DNS
   didn't contain ``priority`` key.
 
   Reported by James Montgomery - @gh-jamesmontgomery.
+
   (GITHUB-1428, GITHUB-1429)
   [Tomaz Muraus]
 

libcloud/storage/base.py

@@ -29,6 +29,7 @@
 import os.path                          # pylint: disable-msg=W0404
 import hashlib
 import warnings
+import errno
 from os.path import join as pjoin
 
 from libcloud.utils.py3 import httplib
@@ -738,7 +739,20 @@ def _hash_buffered_stream(self, stream, hasher, blocksize=65536):
             # Ensure we start from the begining of a stream in case stream is
             # not at the beginning
             if hasattr(stream, 'seek'):
-                stream.seek(0)
+                try:
+                    stream.seek(0)
+                except OSError as e:
+                    if e.errno != errno.ESPIPE:
+                        # This represents "OSError: [Errno 29] Illegal seek"
+                        # error. This could either mean that the underlying
+                        # handle doesn't support seek operation (e.g. pipe) or
+                        # that the invalid seek position is provided. Sadly
+                        # there is no good robust way to distinghuish that so
+                        # we simply ignore all the "Illeal seek" errors so
+                        # this function works correctly with pipes.
+                        # See https://github.com/apache/libcloud/pull/1427 for
+                        # details
+                        raise e
 
             for chunk in libcloud.utils.files.read_in_chunks(iterator=stream):
                 hasher.update(b(chunk))

libcloud/test/storage/test_base.py

@@ -14,6 +14,7 @@
 # limitations under the License.
 
 import sys
+import errno
 import hashlib
 
 from libcloud.utils.py3 import httplib
@@ -92,7 +93,6 @@ def test__upload_object_does_not_stream_response(self):
         response_streamed = mock_response.request.stream
         assert response_streamed is False
 
-
     def test__get_hash_function(self):
         self.driver1.hash_type = 'md5'
         func = self.driver1._get_hash_function()
@@ -141,7 +141,8 @@ def test_upload_no_content_type_supplied_or_detected(self):
     def test_upload_object_hash_calculation_is_efficient(self, mock_read_in_chunks,
                                                          mock_exhaust_iterator):
         # Verify that we don't buffer whole file in memory when calculating
-        # object has when iterator has __next__ method, but instead read and calculate hash in chunks
+        # object has when iterator has __next__ method, but instead read and
+        # calculate hash in chunks
         size = 100
 
         self.driver1.connection = Mock()
@@ -210,6 +211,47 @@ def test_upload_object_hash_calculation_is_efficient(self, mock_read_in_chunks,
         self.assertEqual(mock_read_in_chunks.call_count, 2)
         self.assertEqual(mock_exhaust_iterator.call_count, 0)
 
+    def test_upload_object_via_stream_illegal_seek_errors_are_ignored(self):
+        # Illegal seek errors should be ignored
+        size = 100
+
+        self.driver1.connection = Mock()
+
+        seek_error = OSError('Illegal seek')
+        seek_error.errno = 29
+        assert errno.ESPIPE == 29
+
+        iterator = BodyStream('a' * size)
+        iterator.seek = mock.Mock(side_effect=seek_error)
+
+        result = self.driver1._upload_object(object_name='test1',
+                                             content_type=None,
+                                             request_path='/',
+                                             stream=iterator)
+
+        hasher = hashlib.md5()
+        hasher.update(b('a') * size)
+        expected_hash = hasher.hexdigest()
+
+        self.assertEqual(result['data_hash'], expected_hash)
+        self.assertEqual(result['bytes_transferred'], size)
+
+        # But others shouldn't
+        self.driver1.connection = Mock()
+
+        seek_error = OSError('Other error')
+        seek_error.errno = 21
+
+        iterator = BodyStream('b' * size)
+        iterator.seek = mock.Mock(side_effect=seek_error)
+
+        self.assertRaisesRegex(OSError, 'Other error',
+                               self.driver1._upload_object,
+                               object_name='test1',
+                               content_type=None,
+                               request_path='/',
+                               stream=iterator)
+
 
 if __name__ == '__main__':
     sys.exit(unittest.main())