def _run(self, thread):
     if time.time() - self.heartbeat >= 15:
         self.heartbeat = time.time()
         self._log_status("PUTS")
     name = uuid.uuid4().hex
     if self.object_sources:
         source = random.choice(self.files)
     else:
         source = "0" * self.object_size
     device = random.choice(self.devices)
     partition = str(random.randint(1, 3000))
     container_name = random.choice(self.containers)
     with self.connection() as conn:
         try:
             if self.use_proxy:
                 client.put_object(
                     self.url, self.token, container_name, name, source, content_length=len(source), http_conn=conn
                 )
             else:
                 node = {"ip": self.ip, "port": self.port, "device": device}
                 direct_client.direct_put_object(
                     node, partition, self.account, container_name, name, source, content_length=len(source)
                 )
         except client.ClientException, e:
             self.logger.debug(str(e))
             self.failures += 1

 def _run(self, thread):
     if time.time() - self.heartbeat >= 15:
         self.heartbeat = time.time()
         self._log_status('PUTS')
     name = uuid.uuid4().hex
     if self.object_sources:
         source = random.choice(self.files)
     elif self.upper_object_size > self.lower_object_size:
         source = SourceFile(random.randint(self.lower_object_size,
                                            self.upper_object_size))
     else:
         source = SourceFile(self.object_size)
     device = random.choice(self.devices)
     partition = str(random.randint(1, 3000))
     container_name = random.choice(self.containers)
     with self.connection() as conn:
         try:
             if self.use_proxy:
                 client.put_object(self.url, self.token,
                                   container_name, name, source,
                                   content_length=len(source),
                                   http_conn=conn)
             else:
                 node = {'ip': self.ip, 'port': self.port, 'device': device}
                 direct_client.direct_put_object(node, partition,
                                                 self.account,
                                                 container_name, name,
                                                 source,
                                                 content_length=len(source))
         except client.ClientException as e:
             self.logger.debug(str(e))
             self.failures += 1
         else:
             self.names.append((device, partition, name, container_name))
     self.complete += 1

    def test_direct_put_object_fail(self):
        contents = six.BytesIO(b'123456')

        with mocked_http_conn(500) as conn:
            with self.assertRaises(ClientException) as raised:
                direct_client.direct_put_object(
                    self.node, self.part, self.account, self.container,
                    self.obj, contents)
            self.assertEqual(conn.host, self.node['ip'])
            self.assertEqual(conn.port, self.node['port'])
            self.assertEqual(conn.method, 'PUT')
            self.assertEqual(conn.path, self.obj_path)
        self.assertEqual(raised.exception.http_status, 500)

    def reconstruct_fa(self, conn, target, job):
        """
        Use internal client to rebuild the object data needed to reconstruct
        the fragment archive that is missing on the target node.

        :param conn: a KineticClient connection to the device which has the
                      object metadata for the fragment archive we're rebuilding
        :param target: the ring node which is missing the fragment archive
        :param job: the job dict
        """
        key_info = split_key(job['key'])
        # get object info from conn
        resp = conn.get(job['key'])
        entry = resp.wait()
        info = msgpack.unpackb(entry.value)
        # internal client GET
        account, container, obj = split_path(
            info['name'], 3, rest_with_last=True)
        status, headers, body_iter = self.swift.get_object(
            account, container, obj, {})
        if status // 100 != 2:
            return False
        if headers['x-timestamp'] != key_info['timestamp']:
            return False
        for header in ('etag', 'content-length'):
            headers.pop(header, None)
        headers.update({
            'X-Object-Sysmeta-Ec-Frag-Index': target['index'],
            'X-Backend-Storage-Policy-Index': int(job['policy']),
        })

        # make ec_frag body iter
        def make_segment_iter(segment_size):
            segment_buff = ''
            for chunk in body_iter:
                if not chunk:
                    break
                segment_buff += chunk
                if len(segment_buff) >= segment_size:
                    yield segment_buff[:segment_size]
                    segment_buff = segment_buff[segment_size:]
            if segment_buff:
                yield segment_buff

        def make_frag_iter(policy, frag_index):
            for segment in make_segment_iter(policy.ec_segment_size):
                yield policy.pyeclib_driver.encode(segment)[frag_index]
        # direct client PUT
        direct_put_object(target, job['part'], account, container, obj,
                          make_frag_iter(job['policy'], target['index']),
                          headers=headers)

    def test_direct_put_object_fail(self):
        contents = six.StringIO('123456')

        with mocked_http_conn(500) as conn:
            try:
                direct_client.direct_put_object(
                    self.node, self.part, self.account, self.container,
                    self.obj, contents)
            except ClientException as err:
                pass
            else:
                self.fail('ClientException not raised')
            self.assertEqual(conn.method, 'PUT')
            self.assertEqual(conn.path, self.obj_path)
        self.assertEqual(err.http_status, 500)

    def test_direct_put_object_chunked(self):
        contents = six.StringIO('123456')

        with mocked_http_conn(200) as conn:
            resp = direct_client.direct_put_object(
                self.node, self.part, self.account, self.container, self.obj,
                contents)
            self.assertEqual(conn.method, 'PUT')
            self.assertEqual(conn.path, self.obj_path)
        self.assertEqual(md5('6\r\n123456\r\n0\r\n\r\n').hexdigest(), resp)

    def test_direct_put_object_with_content_length(self):
        contents = StringIO.StringIO('123456')

        with mocked_http_conn(200) as conn:
            resp = direct_client.direct_put_object(
                self.node, self.part, self.account, self.container, self.obj,
                contents, 6)
            self.assertEqual(conn.method, 'PUT')
            self.assertEqual(conn.path, self.obj_path)
        self.assertEqual(md5('123456').hexdigest(), resp)

    def test_direct_put_object_header_content_length(self):
        contents = six.StringIO('123456')
        stub_headers = HeaderKeyDict({
            'Content-Length': '6'})

        with mocked_http_conn(200) as conn:
            resp = direct_client.direct_put_object(
                self.node, self.part, self.account, self.container, self.obj,
                contents, headers=stub_headers)
            self.assertEqual('PUT', conn.method)
            self.assertEqual(conn.req_headers['Content-length'], '6')
        self.assertEqual(md5('123456').hexdigest(), resp)

 def test_direct_put_object_args(self):
     # One test to cover all missing checks
     contents = ""
     with mocked_http_conn(200) as conn:
         resp = direct_client.direct_put_object(
             self.node, self.part, self.account, self.container, self.obj,
             contents, etag="testing-etag", content_type='Text')
         self.assertEqual('PUT', conn.method)
         self.assertEqual(self.obj_path, conn.path)
         self.assertEqual(conn.req_headers['Content-Length'], '0')
         self.assertEqual(conn.req_headers['Content-Type'], 'Text')
     self.assertEqual(md5('0\r\n\r\n').hexdigest(), resp)

    def test_direct_put_object_chunked(self):
        node = {'ip': '1.2.3.4', 'port': '6000', 'device': 'sda'}
        part = '0'
        account = 'a'
        container = 'c'
        name = 'o'
        contents = StringIO.StringIO('123456')

        was_http_connector = direct_client.http_connect
        direct_client.http_connect = mock_http_connect(200)

        resp = direct_client.direct_put_object(node, part, account,
                                               container, name, contents)
        self.assertEqual(md5('6\r\n123456\r\n0\r\n\r\n').hexdigest(), resp)

        direct_client.http_connect = was_http_connector

    def test_reconcile_manifest(self):
        info_url = "%s://%s/info" % (urlparse(self.url).scheme,
                                     urlparse(self.url).netloc)
        proxy_conn = client.http_connection(info_url)
        cluster_info = client.get_capabilities(proxy_conn)
        if 'slo' not in cluster_info:
            raise SkipTest("SLO not enabled in proxy; "
                           "can't test manifest reconciliation")

        # this test is not only testing a split brain scenario on
        # multiple policies with mis-placed objects - it even writes out
        # a static large object directly to the storage nodes while the
        # objects are unavailably mis-placed from *behind* the proxy and
        # doesn't know how to do that for EC_POLICY (clayg: why did you
        # guys let me write a test that does this!?) - so we force
        # wrong_policy (where the manifest gets written) to be one of
        # any of your configured REPL_POLICY (we know you have one
        # because this is a ReplProbeTest)
        wrong_policy = random.choice(POLICIES_BY_TYPE[REPL_POLICY])
        policy = random.choice([p for p in ENABLED_POLICIES
                                if p is not wrong_policy])
        manifest_data = []

        def write_part(i):
            body = 'VERIFY%0.2d' % i + '\x00' * 1048576
            part_name = 'manifest_part_%0.2d' % i
            manifest_entry = {
                "path": "/%s/%s" % (self.container_name, part_name),
                "etag": md5(body).hexdigest(),
                "size_bytes": len(body),
            }
            client.put_object(self.url, self.token, self.container_name,
                              part_name, contents=body)
            manifest_data.append(manifest_entry)

        # get an old container stashed
        self.brain.stop_primary_half()
        self.brain.put_container(int(policy))
        self.brain.start_primary_half()
        # write some parts
        for i in range(10):
            write_part(i)

        self.brain.stop_handoff_half()
        self.brain.put_container(int(wrong_policy))
        # write some more parts
        for i in range(10, 20):
            write_part(i)

        # write manifest
        with self.assertRaises(ClientException) as catcher:
            client.put_object(self.url, self.token, self.container_name,
                              self.object_name,
                              contents=utils.json.dumps(manifest_data),
                              query_string='multipart-manifest=put')

        # so as it works out, you can't really upload a multi-part
        # manifest for objects that are currently misplaced - you have to
        # wait until they're all available - which is about the same as
        # some other failure that causes data to be unavailable to the
        # proxy at the time of upload
        self.assertEqual(catcher.exception.http_status, 400)

        # but what the heck, we'll sneak one in just to see what happens...
        direct_manifest_name = self.object_name + '-direct-test'
        object_ring = POLICIES.get_object_ring(wrong_policy.idx, '/etc/swift')
        part, nodes = object_ring.get_nodes(
            self.account, self.container_name, direct_manifest_name)
        container_part = self.container_ring.get_part(self.account,
                                                      self.container_name)

        def translate_direct(data):
            return {
                'hash': data['etag'],
                'bytes': data['size_bytes'],
                'name': data['path'],
            }
        direct_manifest_data = map(translate_direct, manifest_data)
        headers = {
            'x-container-host': ','.join('%s:%s' % (n['ip'], n['port']) for n
                                         in self.container_ring.devs),
            'x-container-device': ','.join(n['device'] for n in
                                           self.container_ring.devs),
            'x-container-partition': container_part,
            'X-Backend-Storage-Policy-Index': wrong_policy.idx,
            'X-Static-Large-Object': 'True',
        }
        for node in nodes:
            direct_client.direct_put_object(
                node, part, self.account, self.container_name,
                direct_manifest_name,
                contents=utils.json.dumps(direct_manifest_data),
                headers=headers)
            break  # one should do it...

        self.brain.start_handoff_half()
        self.get_to_final_state()
        Manager(['container-reconciler']).once()
        # clear proxy cache
        client.post_container(self.url, self.token, self.container_name, {})
#.........这里部分代码省略.........

    def test_reconcile_manifest(self):
        manifest_data = []

        def write_part(i):
            body = 'VERIFY%0.2d' % i + '\x00' * 1048576
            part_name = 'manifest_part_%0.2d' % i
            manifest_entry = {
                "path": "/%s/%s" % (self.container_name, part_name),
                "etag": md5(body).hexdigest(),
                "size_bytes": len(body),
            }
            client.put_object(self.url, self.token, self.container_name,
                              part_name, contents=body)
            manifest_data.append(manifest_entry)

        # get an old container stashed
        self.brain.stop_primary_half()
        policy = random.choice(ENABLED_POLICIES)
        self.brain.put_container(policy.idx)
        self.brain.start_primary_half()
        # write some parts
        for i in range(10):
            write_part(i)

        self.brain.stop_handoff_half()
        wrong_policy = random.choice([p for p in ENABLED_POLICIES
                                      if p is not policy])
        self.brain.put_container(wrong_policy.idx)
        # write some more parts
        for i in range(10, 20):
            write_part(i)

        # write manifest
        try:
            client.put_object(self.url, self.token, self.container_name,
                              self.object_name,
                              contents=utils.json.dumps(manifest_data),
                              query_string='multipart-manifest=put')
        except ClientException as err:
            # so as it works out, you can't really upload a multi-part
            # manifest for objects that are currently misplaced - you have to
            # wait until they're all available - which is about the same as
            # some other failure that causes data to be unavailable to the
            # proxy at the time of upload
            self.assertEqual(err.http_status, 400)

        # but what the heck, we'll sneak one in just to see what happens...
        direct_manifest_name = self.object_name + '-direct-test'
        object_ring = POLICIES.get_object_ring(wrong_policy.idx, '/etc/swift')
        part, nodes = object_ring.get_nodes(
            self.account, self.container_name, direct_manifest_name)
        container_part = self.container_ring.get_part(self.account,
                                                      self.container_name)

        def translate_direct(data):
            return {
                'hash': data['etag'],
                'bytes': data['size_bytes'],
                'name': data['path'],
            }
        direct_manifest_data = map(translate_direct, manifest_data)
        headers = {
            'x-container-host': ','.join('%s:%s' % (n['ip'], n['port']) for n
                                         in self.container_ring.devs),
            'x-container-device': ','.join(n['device'] for n in
                                           self.container_ring.devs),
            'x-container-partition': container_part,
            'X-Backend-Storage-Policy-Index': wrong_policy.idx,
            'X-Static-Large-Object': 'True',
        }
        for node in nodes:
            direct_client.direct_put_object(
                node, part, self.account, self.container_name,
                direct_manifest_name,
                contents=utils.json.dumps(direct_manifest_data),
                headers=headers)
            break  # one should do it...

        self.brain.start_handoff_half()
        get_to_final_state()
        Manager(['container-reconciler']).once()
        # clear proxy cache
        client.post_container(self.url, self.token, self.container_name, {})

        # let's see how that direct upload worked out...
        metadata, body = client.get_object(
            self.url, self.token, self.container_name, direct_manifest_name,
            query_string='multipart-manifest=get')
        self.assertEqual(metadata['x-static-large-object'].lower(), 'true')
        for i, entry in enumerate(utils.json.loads(body)):
            for key in ('hash', 'bytes', 'name'):
                self.assertEquals(entry[key], direct_manifest_data[i][key])
        metadata, body = client.get_object(
            self.url, self.token, self.container_name, direct_manifest_name)
        self.assertEqual(metadata['x-static-large-object'].lower(), 'true')
        self.assertEqual(int(metadata['content-length']),
                         sum(part['size_bytes'] for part in manifest_data))
        self.assertEqual(body, ''.join('VERIFY%0.2d' % i + '\x00' * 1048576
                                       for i in range(20)))

#.........这里部分代码省略.........

    def test_ec_handoff_duplicate_available(self):
        container_name = 'container-%s' % uuid4()
        object_name = 'object-%s' % uuid4()

        # create EC container
        headers = {'X-Storage-Policy': self.policy.name}
        client.put_container(self.url, self.token, container_name,
                             headers=headers)

        # get our node lists
        opart, onodes = self.object_ring.get_nodes(
            self.account, container_name, object_name)

        # find both primary servers that have both of their devices in
        # the primary node list
        group_nodes_by_config = defaultdict(list)
        for n in onodes:
            group_nodes_by_config[self.config_number(n)].append(n)
        double_disk_primary = []
        for config_number, node_list in group_nodes_by_config.items():
            if len(node_list) > 1:
                double_disk_primary.append((config_number, node_list))

        # sanity, in a 4+2 with 8 disks two servers will be doubled
        self.assertEqual(len(double_disk_primary), 2)

        # shutdown the first double primary
        primary0_config_number, primary0_node_list = double_disk_primary[0]
        Manager(['object-server']).stop(number=primary0_config_number)

        # PUT object
        contents = Body()
        client.put_object(self.url, self.token, container_name,
                          object_name, contents=contents)

        # sanity fetch two frags on handoffs
        handoff_frags = []
        for node in self.object_ring.get_more_nodes(opart):
            headers, data = direct_client.direct_get_object(
                node, opart, self.account, container_name, object_name,
                headers={'X-Backend-Storage-Policy-Index': int(self.policy)}
            )
            handoff_frags.append((node, headers, data))

        # bring the first double primary back, and fail the other one
        Manager(['object-server']).start(number=primary0_config_number)
        primary1_config_number, primary1_node_list = double_disk_primary[1]
        Manager(['object-server']).stop(number=primary1_config_number)

        # we can still GET the object
        resp_etag = self.get_object(container_name, object_name)
        self.assertEqual(resp_etag, contents.etag)

        # now start to "revert" the first handoff frag
        node = primary0_node_list[0]
        handoff_node, headers, data = handoff_frags[0]
        # N.B. object server api returns quoted ETag
        headers['ETag'] = headers['Etag'].strip('"')
        headers['X-Backend-Storage-Policy-Index'] = int(self.policy)
        direct_client.direct_put_object(
            node, opart,
            self.account, container_name, object_name,
            contents=data, headers=headers)

        # sanity - check available frags
        frag2count = self._check_nodes(opart, onodes,
                                       container_name, object_name)
        # ... five frags total
        self.assertEqual(sum(frag2count.values()), 5)
        # ... only 4 unique indexes
        self.assertEqual(len(frag2count), 4)

        # we can still GET the object
        resp_etag = self.get_object(container_name, object_name)
        self.assertEqual(resp_etag, contents.etag)

        # ... but we need both handoffs or we get a error
        for handoff_node, hdrs, data in handoff_frags:
            Manager(['object-server']).stop(
                number=self.config_number(handoff_node))
            with self.assertRaises(Exception) as cm:
                self.get_object(container_name, object_name)
            self.assertIn(cm.exception.http_status, (404, 503))
            Manager(['object-server']).start(
                number=self.config_number(handoff_node))

        # fix everything
        Manager(['object-server']).start(number=primary1_config_number)
        Manager(["object-reconstructor"]).once()

        # sanity - check available frags
        frag2count = self._check_nodes(opart, onodes,
                                       container_name, object_name)
        # ... six frags total
        self.assertEqual(sum(frag2count.values()), 6)
        # ... all six unique
        self.assertEqual(len(frag2count), 6)