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

问题导读

1、如何运行守护进程程序？
2、一次container审计操作的流程是怎样的？
3、怎样对以.db为后缀的文件生成元组？









概述部分：
容器审计守护进程；  
审计的主要的职责就是检查容器数据是否被删除（通过解析账户目录下的.db文件）；
这里定义的once=True，说明系统默认调用守护进程类Daemon中的run_once方法，有人说run_once主要应用于测试，这个没有验证；
从而最终实现调用ContainerAuditor类中的run_once方法；
如果调用的是ContainerAuditor类中的run_forever方法，则会实现循环实现检查指定容器数据是否被删除；

源码解析部分：

from swift.container.auditor import ContainerAuditor
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(ContainerAuditor, conf_file, **options)
def run_daemon(klass, conf_file, section_name='', once=False, **kwargs):
    """
    从配置文件加载设置，然后实例化守护进程“klass”并运行这个守护进程通过指定的参数kwarg；
    """
    ......
   try:
       klass(conf).run(once=once, **kwargs)
   except KeyboardInterrupt:
       logger.info('User quit')
   logger.info('Exited')
def run(self, once=False, **kwargs):
    """
   Run the daemon
    运行守护进程程序；
    即运行方法run_once，或者方法run_forever；
    """
    # 配置参数相关；
   utils.validate_configuration()
   utils.drop_privileges(self.conf.get('user', 'swift'))
    # 日志相关处理；
   utils.capture_stdio(self.logger, **kwargs)
   def kill_children(*args):
       signal.signal(signal.SIGTERM, signal.SIG_IGN)
       os.killpg(0, signal.SIGTERM)
       sys.exit()
   signal.signal(signal.SIGTERM, kill_children)
   if once:
       self.run_once(**kwargs)
   else:
       self.run_forever(**kwargs)
def run_once(self, *args, **kwargs):
    """
   Override this to run the script once
    子类中的方法需要被重写；
    """
   raise NotImplementedError('run_once not implemented')
def run_once(self, *args, **kwargs):
    """
    执行一次container审计操作；
    """
   self.logger.info(_('Begin container audit "once" mode'))
   begin = reported = time.time()    
    # 进行一次审计操作；
   self._one_audit_pass(reported)
   elapsed = time.time() - begin
   self.logger.info(
        ('Container audit "once" mode completed: %.02fs'), elapsed)
        dump_recon_cache({'container_auditor_pass_completed': elapsed},
                    self.rcache, self.logger)
def _one_audit_pass(self, reported):
    """
    对container进行审计操作；
    """
    # 基于给定的设备路径和文件后缀，为在这个数据目录中的所有以.db为后缀的文件生成元组（path, device, partition）；
    # 通过方法audit_location_generator方法计算获得path，device，partition；
   all_locs = audit_location_generator(self.devices, DATADIR, '.db',
                                       mount_check=self.mount_check,
                                       logger=self.logger)
   for path, device, partition in all_locs:
       # container_audit：对给定的container路径的db文件进行检查；
          # 审计给定的容器，看看是否已经被删除，如果没有删除，则为容器获取全局数据；
       self.container_audit(path)
       if time.time() - reported >= 3600:  # once an hour
           self.logger.info(_('Since %(time)s: Container audits: %(pass)s passed '
                              'audit, %(fail)s failed audit'),
                            {'time': time.ctime(reported),
                             'pass': self.container_passes,
                             'fail': self.container_failures})
           dump_recon_cache(
                            {'container_audits_since': reported,
                             'container_audits_passed': self.container_passes,
                             'container_audits_failed': self.container_failures},
                             self.rcache, self.logger)
           reported = time.time()
           self.container_passes = 0
           self.container_failures = 0
       self.containers_running_time = ratelimit_sleep(self.containers_running_time, self.max_containers_per_second)
复制代码

    return reported1.调用方法audit_location_generator，实现基于给定的设备路径（self.devices，DATADIR=containers）和文件后缀（.db），为在这个数据目录中的所有以.db为后缀的文件生成元组（path, device, partition）；
生成路径示例如下：

path = /devices/device/containers/partition/asuffix/hsh/****.db
device：self.devices下具体的一个设备名称；
partition：/devices/device/containers/下具体的一个分区名称；
复制代码

2.遍历元组（path, device, partition）中的每一个匹配对，调用方法container_audit实现审计给定的容器，看看是否已经被删除，如果没有删除，则为账户获取全局数据；

转到2，来看方法container_audit：

def container_audit(self, path):
    """
    对给定的container路径的db文件进行检查；
    审计给定的容器，看看是否被视为删除，如果没有删除，则为容器获取全局数据；
    """
   start_time = time.time()
   try:
       # ContainerBroker：container数据库的封装工作；
       broker = ContainerBroker(path)
       # is_deleted：检测指定容器的数据库是否被视为删除；
       if not broker.is_deleted():
           # get_info：获取container的全局数据；
           broker.get_info()
           self.logger.increment('passes')
           self.container_passes += 1
           self.logger.debug(_('Audit passed for %s'), broker)
       except (Exception, Timeout):
           self.logger.increment('failures')
           self.container_failures += 1
           self.logger.exception(_('ERROR Could not get container info %s'), path)
       self.logger.timing_since('timing', start_time)这个方法的实现比较简单，这里就不再进一步进行解析。
复制代码