Merge "Set backend content length for fallocate - EC Policy"

swift/proxy/controllers/obj.py

@@ -1772,7 +1772,7 @@ def _send_file(self, write_timeout, exception_handler):
 
     @classmethod
     def connect(cls, node, part, path, headers, conn_timeout, node_timeout,
-                chunked=False):
+                chunked=False, expected_frag_archive_size=None):
         """
         Connect to a backend node and send the headers.
 
@@ -1794,9 +1794,10 @@ def connect(cls, node, part, path, headers, conn_timeout, node_timeout,
         # we must use chunked encoding.
         headers['Transfer-Encoding'] = 'chunked'
         headers['Expect'] = '100-continue'
-        if 'Content-Length' in headers:
-            headers['X-Backend-Obj-Content-Length'] = \
-                headers.pop('Content-Length')
+
+        # make sure this isn't there
+        headers.pop('Content-Length')
+        headers['X-Backend-Obj-Content-Length'] = expected_frag_archive_size
 
         headers['X-Backend-Obj-Multipart-Mime-Boundary'] = mime_boundary
 
@@ -2121,16 +2122,41 @@ def _connect_put_node(self, node_iter, part, path, headers,
         # the object server will get different bytes, so these
         # values do not apply (Content-Length might, in general, but
         # in the specific case of replication vs. EC, it doesn't).
-        headers.pop('Content-Length', None)
+        client_cl = headers.pop('Content-Length', None)
         headers.pop('Etag', None)
 
+        expected_frag_size = None
+        if client_cl:
+            policy_index = int(headers.get('X-Backend-Storage-Policy-Index'))
+            policy = POLICIES.get_by_index(policy_index)
+            # TODO: PyECLib <= 1.2.0 looks to return the segment info
+            # different from the input for aligned data efficiency but
+            # Swift never does. So calculate the fragment length Swift
+            # will actually send to object sever by making two different
+            # get_segment_info calls (until PyECLib fixed).
+            # policy.fragment_size makes the call using segment size,
+            # and the next call is to get info for the last segment
+
+            # get number of fragments except the tail - use truncation //
+            num_fragments = int(client_cl) // policy.ec_segment_size
+            expected_frag_size = policy.fragment_size * num_fragments
+
+            # calculate the tail fragment_size by hand and add it to
+            # expected_frag_size
+            last_segment_size = int(client_cl) % policy.ec_segment_size
+            if last_segment_size:
+                last_info = policy.pyeclib_driver.get_segment_info(
+                    last_segment_size, policy.ec_segment_size)
+                expected_frag_size += last_info['fragment_size']
+
         self.app.logger.thread_locals = logger_thread_locals
         for node in node_iter:
             try:
                 putter = ECPutter.connect(
                     node, part, path, headers,
                     conn_timeout=self.app.conn_timeout,
-                    node_timeout=self.app.node_timeout)
+                    node_timeout=self.app.node_timeout,
+                    expected_frag_archive_size=expected_frag_size)
                 self.app.set_node_timing(node, putter.connect_duration)
                 return putter
             except InsufficientStorage:

test/unit/proxy/controllers/test_obj.py

@@ -1497,6 +1497,8 @@ def capture_headers(ip, port, device, part, method, path, headers,
             conn_id = kwargs['connection_id']
             put_requests[conn_id]['boundary'] = headers[
                 'X-Backend-Obj-Multipart-Mime-Boundary']
+            put_requests[conn_id]['backend-content-length'] = headers[
+                'X-Backend-Obj-Content-Length']
 
         with set_http_connect(*codes, expect_headers=expect_headers,
                               give_send=capture_body,
@@ -1510,6 +1512,9 @@ def capture_headers(ip, port, device, part, method, path, headers,
             self.assertTrue(info['boundary'] is not None,
                             "didn't get boundary for conn %r" % (
                                 connection_id,))
+            self.assertTrue(size > int(info['backend-content-length']) > 0,
+                            "invalid backend-content-length for conn %r" % (
+                                connection_id,))
 
             # email.parser.FeedParser doesn't know how to take a multipart
             # message and boundary together and parse it; it only knows how
@@ -1531,6 +1536,13 @@ def capture_headers(ip, port, device, part, method, path, headers,
             self.assertEqual(obj_part['X-Document'], 'object body')
             frag_archives.append(obj_part.get_payload())
 
+            # assert length was correct for this connection
+            self.assertEqual(int(info['backend-content-length']),
+                             len(frag_archives[-1]))
+            # assert length was the same for all connections
+            self.assertEqual(int(info['backend-content-length']),
+                             len(frag_archives[0]))
+
             # validate some footer metadata
             self.assertEqual(footer_part['X-Document'], 'object metadata')
             footer_metadata = json.loads(footer_part.get_payload())