Swift源码分析----swift-account-reaper(2)

感谢朋友支持本博客，欢迎共同探讨交流，由于能力和时间有限，错误之处在所难免，欢迎指正！

如果转载，请保留作者信息。

博客地址：http://blog.csdn.net/gaoxingnengjisuan

邮箱地址：[email protected]

PS：最近没有登录博客，很多朋友的留言没有看见，这里道歉！还有就是本人较少上QQ，可以邮件交流。

接续上一篇博客：

def reap_container(self, account, account_partition, account_nodes, container):
     """        
     实现收割container操作；
     实现删除容器container下数据和容器container本身；
     当执行删除一个单独的object出现异常的时候，进程将会继续执行删除container中其它object的操作，
     删除失败的object将会在下一次这个方法被调用的时候继续尝试删除操作；
     当获取要删除的object列表出现异常的时候，进程将会停止（但是也将会在下一次这个方法被调用的时候继续尝试这个操作）；
     如果所有object已经被删除，将会通过发送一个删除请求到所有的container节点，来实现删除container本身；
     每个container服务的删除将会更新对应的account服务，且从account列表删除container；
     """
    account_nodes = list(account_nodes)
        
    # get_container_ring：获取swift.common.ring.Ring对象，名称为'container'；
    # 为account/container/object获取分区和节点信息；
    # 返回元组（分区，节点信息列表）；
    # 在节点信息列表中至少包含id、weight、zone、ip、port、device、meta；
    part, nodes = self.get_container_ring().get_nodes(account, container)
    node = nodes[-1]
    # 绿色线程的连接池类；
    pool = GreenPool(size=self.object_concurrency)
    marker = ''
        
    # 删除从container中获取的所有object；
    while True:
        objects = None
        try:
            # direct_get_container：发送调用'GET'方法的请求，实现从容器服务（器）直接获取容器中内容的列表；
            # objects获取所哟的object；
            objects = direct_get_container(
                        node, part, account, container,
                        marker=marker,
                        conn_timeout=self.conn_timeout,
                        response_timeout=self.node_timeout)[1]
            self.stats_return_codes[2] = self.stats_return_codes.get(2, 0) + 1
            self.logger.increment('return_codes.2')
        except ClientException as err:
            if self.logger.getEffectiveLevel() <= DEBUG:
                self.logger.exception(_('Exception with %(ip)s:%(port)s/%(device)s'), node)
            self.stats_return_codes[err.http_status / 100] = self.stats_return_codes.get(err.http_status / 100, 0) + 1
            self.logger.increment('return_codes.%d' % (err.http_status / 100,))
        if not objects:
            break
            
        # 删除从container中获取的所有object；
        try:
            for obj in objects:
                if isinstance(obj['name'], unicode):
                    obj['name'] = obj['name'].encode('utf8')
                    
                # 在绿色线程中执行方法reap_object；
                # reap_object：通过一个发送到每个节点的删除object请求，执行删除给定的obj['name']；
                # 执行删除请求，每个object服务将会更新删除相应的容器服务器；
                # 并从container列表删除这个object；
                pool.spawn(self.reap_object, account, container, part, nodes, obj['name'])
            pool.waitall()
        except (Exception, Timeout):
            self.logger.exception(_('Exception with objects for container '
                                     '%(container)s for account %(account)s'),
                                   {'container': container, 'account': account})
        marker = objects[-1]['name']
        if marker == '':
            break
    successes = 0
    failures = 0
        
    # 在所有相关节点中，删除container的相关信息；
    for node in nodes:
        anode = account_nodes.pop()
        try:
            # direct_delete_container：发送调用'DELETE'方法的请求，实现从account服务（器）直接删除container；
            direct_delete_container(
                    node, part, account, container,
                    conn_timeout=self.conn_timeout,
                    response_timeout=self.node_timeout,
                    headers={'X-Account-Host': '%(ip)s:%(port)s' % anode,
                             'X-Account-Partition': str(account_partition),
                             'X-Account-Device': anode['device'],
                             'X-Account-Override-Deleted': 'yes'})

            successes += 1
            self.stats_return_codes[2] = self.stats_return_codes.get(2, 0) + 1
            self.logger.increment('return_codes.2')
        except ClientException as err:
            if self.logger.getEffectiveLevel() <= DEBUG:
                self.logger.exception(_('Exception with %(ip)s:%(port)s/%(device)s'), node)
            failures += 1
            self.logger.increment('containers_failures')
            self.stats_return_codes[err.http_status / 100] = self.stats_return_codes.get(err.http_status / 100, 0) + 1
            self.logger.increment('return_codes.%d' % (err.http_status / 100,))
    if successes > failures:
        self.stats_containers_deleted += 1
        self.logger.increment('containers_deleted')
    elif not successes:
        self.stats_containers_remaining += 1
        self.logger.increment('containers_remaining')
    else:
        self.stats_containers_possibly_remaining += 1
        self.logger.increment('containers_possibly_remaining')
           

1.获取指定account指定container的分区号和所有副本所在节点信息，从节点列表中获取一个节点（因为所有节点上的副本信息都是一致的）；

2.调用方法direct_get_container，实现发送调用'GET'方法的请求，实现从容器服务（器）直接获取指定容器下的所有对象列表；

3.遍历所有的对象，针对每一个对象调用方法reap_object实现从对象服务器删除指定对象的数据信息；

4.遍历container的所有副本所在节点，针对每一个节点调用方法direct_delete_container实现从服务（器）直接删除container相关数据信息（比如元数据信息数据库信息等等）；

转到3，来看方法reap_object的实现：

def reap_object(self, account, container, container_partition, container_nodes, obj):
     """        
     实现收割object操作；
     通过一个发送到每个节点的删除object请求，执行删除给定的object；
     执行删除请求，每个object服务将会更新删除相应的容器服务器；
     并从container列表删除这个object；
     """
    # container副本相关节点；
    container_nodes = list(container_nodes)
        
    # 为account/container/object获取分区和节点信息；
    # 返回元组（分区，节点信息列表）；
    # 在节点信息列表中至少包含id、weight、zone、ip、port、device、meta；
    # get_object_ring：获取swift.common.ring.Ring对象，名称为'object'；        
    part, nodes = self.get_object_ring().get_nodes(account, container, obj)
    successes = 0
    failures = 0
    for node in nodes:
        cnode = container_nodes.pop()
        try:
            # 建立一个HTTPConnection类的对象；
            # 发出的HTTP请求的方法'DELETE'到服务器；
            # 直接从对象服务（器）删除对象；
            # 获取来自服务器的响应；
            direct_delete_object(
                    node, part, account, container, obj,
                    conn_timeout=self.conn_timeout,
                    response_timeout=self.node_timeout,
                    headers={'X-Container-Host': '%(ip)s:%(port)s' % cnode,
                             'X-Container-Partition': str(container_partition),
                             'X-Container-Device': cnode['device']})

            successes += 1
            self.stats_return_codes[2] = self.stats_return_codes.get(2, 0) + 1
            self.logger.increment('return_codes.2')
        except ClientException as err:
            if self.logger.getEffectiveLevel() <= DEBUG:
                self.logger.exception(_('Exception with %(ip)s:%(port)s/%(device)s'), node)
            failures += 1
            self.logger.increment('objects_failures')
            self.stats_return_codes[err.http_status / 100] = self.stats_return_codes.get(err.http_status / 100, 0) + 1
            self.logger.increment('return_codes.%d' % (err.http_status / 100,))
        if successes > failures:
            self.stats_objects_deleted += 1
            self.logger.increment('objects_deleted')
        elif not successes:
            self.stats_objects_remaining += 1
            self.logger.increment('objects_remaining')
        else:
            self.stats_objects_possibly_remaining += 1
            self.logger.increment('objects_possibly_remaining')
           

3.1.获取指定account指定container指定object的分区号和所有副本所在节点信息；

3.2.遍历所有副本所在节点，针对每一个节点调用方法direct_delete_object实现发送调用'DELETE'方法的请求，实现从对象服务（器）直接删除指定对象数据；

转到3.2，来看方法direct_delete_object的实现：

def direct_delete_object(node, part, account, container, obj, conn_timeout=5, response_timeout=15, headers=None):
     """
     直接从对象服务（器）删除对象；
     建立一个HTTPConnection类的对象；
     发出的HTTP请求的方法'DELETE'到服务器；
     直接从对象服务（器）删除对象；
     获取来自服务器的响应；
     """
    if headers is None:
        headers = {}

    path = '/%s/%s/%s' % (account, container, obj)
    with Timeout(conn_timeout):
        # 建立一个HTTPConnection类的对象；
        # 发出的HTTP请求的方法'DELETE'；
        # 返回HTTPConnection连接对象；
        conn = http_connect(node['ip'], node['port'], node['device'], part,
                            'DELETE', path, headers=gen_headers(headers, True))
    with Timeout(response_timeout):
        # getresponse：获取来自服务器的响应；
        resp = conn.getresponse()
        resp.read()
    if not is_success(resp.status):
        raise ClientException(
            'Object server %s:%s direct DELETE %s gave status %s' %
            (node['ip'], node['port'],
             repr('/%s/%s%s' % (node['device'], part, path)),
             resp.status),
            http_host=node['ip'], http_port=node['port'],
            http_device=node['device'], http_status=resp.status,
            http_reason=resp.reason)
           

转到4，来看方法direct_delete_container的实现：

def direct_delete_container(node, part, account, container, conn_timeout=5, response_timeout=15, headers=None):
     """
     发送调用'DELETE'方法的请求，实现从容器服务（器）直接删除container相关数据；
     """
    if headers is None:
        headers = {}

    path = '/%s/%s' % (account, container)
    with Timeout(conn_timeout):
        conn = http_connect(node['ip'], node['port'], node['device'], part,
                            'DELETE', path, headers=gen_headers(headers, True))
    with Timeout(response_timeout):
        resp = conn.getresponse()
        resp.read()
    if not is_success(resp.status):
        raise ClientException(
            'Container server %s:%s direct DELETE %s gave status %s' %
            (node['ip'], node['port'],
             repr('/%s/%s%s' % (node['device'], part, path)), resp.status),
            http_host=node['ip'], http_port=node['port'],
            http_device=node['device'], http_status=resp.status,
            http_reason=resp.reason)