天天看点
返回

Swift源码分析----swift-container-updater

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

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

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

邮箱地址：[email protected]

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

概述部分：

容器数据更新守护进程；

遍历所有设备下所有容器下的DB文件，

如果DB文件指定的容器发生数据更新，

通过HTTP协议应用PUT方法实现报告container信息给相应的account服务，

在account中实现相应的更新信息操作；

源码解析部分：

下面是这部分代码的主要执行流程，代码中较重要的部分已经进行了相关的注释；

from swift.container.updater import ContainerUpdater
from swift.common.utils import parse_options
from swift.common.daemon import run_daemon

if __name__ == '__main__':
    conf_file, options = parse_options(once=True)
    run_daemon(ContainerUpdater, conf_file, **options)
           
def run_once(self, *args, **kwargs):
     """
     遍历所有设备下所有容器下的DB文件，如果DB文件指定的容器发生数据更新，
     通过HTTP协议应用PUT方法实现报告container信息给相应的account服务，
     在account中实现相应的更新信息操作；
     """
    patcher.monkey_patch(all=False, socket=True)
    self.logger.info(_('Begin container update single threaded sweep'))
    begin = time.time()
    self.no_changes = 0
    self.successes = 0
    self.failures = 0
        
    # get_paths：获取所有设备下容器下的所有分区的具体路径；
      # 遍历所有容器相关的所有分区的具体路径；
    for path in self.get_paths():
           # 遍历path下所有文件（树形遍历），查找DB文件，并调用方法process_container对文件进行处理；
           # 如果DB文件指定的容器发生数据更新，通过HTTP协议应用PUT方法实现报告container信息给相应的account服务；
           # 方法process_container实现了对container进行处理，并更新它在对应account中的信息；
        self.container_sweep(path)
    elapsed = time.time() - begin
    self.logger.info(_(
        'Container update single threaded sweep completed: '
        '%(elapsed).02fs, %(success)s successes, %(fail)s failures, '
        '%(no_change)s with no changes'),
        {'elapsed': elapsed, 'success': self.successes,
         'fail': self.failures, 'no_change': self.no_changes})
    dump_recon_cache({'container_updater_sweep': elapsed},
                     self.rcache, self.logger)

1.获取所有设备下容器(containers)下的所有分区的具体路径；

2.遍历所有分区的具体路径，调用方法container_sweep实现获取分区下的DB文件，如果DB文件指定的容器发生数据更新，通过HTTP协议应用PUT方法实现报告container信息给相应的account服务，在account中实现相应的更新信息操作；

转到2，来看方法container_sweep的实现：

def container_sweep(self, path):
    """
    遍历path下所有文件（树形遍历），查找DB文件，并调用方法process_container对文件进行处理；
    如果DB文件指定的容器发生数据更新，通过HTTP协议应用PUT方法实现报告container信息给相应的account服务；
    方法process_container实现了对container进行处理，并更新它在对应account中的信息；
    """
    for root, dirs, files in os.walk(path):
        for file in files:
            if file.endswith('.db'):
                # process_container：对container进行处理，并更新它在对应account中的信息；
                # 如果数据库文件指定的容器发生数据更新，通过HTTP协议应用PUT方法实现报告container信息给相应的account服务；
                self.process_container(os.path.join(root, file))
                time.sleep(self.slowdown)

查找给定分区目录下所有的以.db为后缀的文件，并调用方法process_container实现对container进行处理，并更新它在对应account中的信息；如果数据库文件指定的容器发生数据更新，通过HTTP协议应用PUT方法实现报告container信息给相应的account服务；

来看方法process_container的实现：

def process_container(self, dbfile):
    """
    对container进行处理，并更新它在对应account中的信息；
    如果容器发生数据更新，通过HTTP协议应用PUT方法实现报告container信息给相应的account服务；
    """
    start_time = time.time()
        
    # 容器数据库的封装类实现；
    broker = ContainerBroker(dbfile, logger=self.logger)
    # get_info：获取container的全局数据；
    # 包括account/container/created_at/put_timestamp/delete_timestamp/object_count/bytes_used/reported_put_timestamp/reported_delete_timestamp/reported_object_count/reported_bytes_used/hash/id等信息；
    info = broker.get_info()
    # Don't send updates if the container was auto-created since it
    # definitely doesn't have up to date statistics.
    if float(info['put_timestamp']) <= 0:
        return
        
    if self.account_suppressions.get(info['account'], 0) > time.time():
        return
        
    # 如果满足下述条件，说明容器发生数据更新；
    # 通过HTTP协议应用PUT方法实现报告container信息给account服务；
    if info['put_timestamp'] > info['reported_put_timestamp'] or info['delete_timestamp'] > info['reported_delete_timestamp'] \
       or info['object_count'] != info['reported_object_count'] or info['bytes_used'] != info['reported_bytes_used']:
        container = '/%s/%s' % (info['account'], info['container'])
            
        # 获取容器对应的account的相关分区和节点信息；
        # 返回元组（分区，节点信息列表）；
        # 在节点信息列表中至少包含id、weight、zone、ip、port、device、meta；
        part, nodes = self.get_account_ring().get_nodes(info['account'])
            
        # 在绿色线程中执行方法container_report，实现报告container信息给account服务；
        # container_report：通过HTTP协议应用PUT方法实现报告container信息给account服务；
        events = [spawn(self.container_report, node, part, container,
                        info['put_timestamp'], info['delete_timestamp'],
                        info['object_count'], info['bytes_used'])
                  for node in nodes]
            
        successes = 0
        for event in events:
            if is_success(event.wait()):
                successes += 1
        if successes >= quorum_size(len(events)):
            self.logger.increment('successes')
            self.successes += 1
            self.logger.debug(_('Update report sent for %(container)s %(dbfile)s'),
                               {'container': container, 'dbfile': dbfile})
            # reported：更新数据库中container_stat表的reported_put_timestamp、reported_delete_timestamp、reported_object_count和reported_bytes_used；
            broker.reported(info['put_timestamp'],
                            info['delete_timestamp'], info['object_count'],
                            info['bytes_used'])
        else:
            self.logger.increment('failures')
            self.failures += 1
            self.logger.debug(_('Update report failed for %(container)s %(dbfile)s'), {'container': container, 'dbfile': dbfile})
            self.account_suppressions[info['account']] = until = \
                time.time() + self.account_suppression_time
            if self.new_account_suppressions:
                print >>self.new_account_suppressions, info['account'], until
        # Only track timing data for attempted updates:
        self.logger.timing_since('timing', start_time)
    else:
        self.logger.increment('no_changes')
        self.no_changes += 1

2.1.根据指定的数据库文件（后缀为.db的文件），实现获取container的全局数据信息；

2.2.根据container的全局数据信息中若干属性的比较，判断该容器是否发生了数据的更新；

2.3 如果容器发生了数据更新，调用方法get_nodes获取容器所属account的分区号和副本节点；

2.4.针对容器所属account的每一个副本节点，调用方法container_report实现通过HTTP协议应用PUT方法实现报告container信息给account服务；

注：这里就是处理一个分区上所有的容器，首先验证容器是否发生数据更新，如果容器发生数据更新，则先获取该容器所属账户的分区，和所有账户的副本节点，再遍历每个副本节点，实现把容器数据更新的信息报告给账户的每一个副本节点；

转到2.4，来看方法container_report的实现：

def container_report(self, node, part, container, put_timestamp, delete_timestamp, count, bytes):
    """
    通过HTTP协议应用PUT方法实现报告container信息给account服务；
    """
    with ConnectionTimeout(self.conn_timeout):
        try:
            headers = {
                'X-Put-Timestamp': put_timestamp,
                'X-Delete-Timestamp': delete_timestamp,
                'X-Object-Count': count,
                'X-Bytes-Used': bytes,
                'X-Account-Override-Deleted': 'yes',
                'user-agent': self.user_agent}
                
            # 通过HTTP协议应用PUT方法实现报告container信息给account服务；
            # 建立一个HTTPConnection类的对象；
            # 发出的HTTP请求的方法'PUT'；
            # 返回HTTPConnection连接对象；
            conn = http_connect(node['ip'], node['port'], node['device'], part, 'PUT', container, headers=headers)
        except (Exception, Timeout):
            self.logger.exception(_(
                'ERROR account update failed with '
                '%(ip)s:%(port)s/%(device)s (will retry later): '), node)
            return HTTP_INTERNAL_SERVER_ERROR
    with Timeout(self.node_timeout):
        try:
            # 获取来自服务器的响应；
            resp = conn.getresponse()
            resp.read()
            return resp.status
        except (Exception, Timeout):
            if self.logger.getEffectiveLevel() <= logging.DEBUG:
                self.logger.exception(_('Exception with %(ip)s:%(port)s/%(device)s'), node)
            return HTTP_INTERNAL_SERVER_ERROR
        finally:
            conn.close()
OpenStack源码分析-SWIFT OpenStack源码分析 openstack swift 源代码
上一篇: Swift源码分析----swift-object-expirer
下一篇: Swift源码分析----swift-object-updater

继续阅读