本文所使用的jumpserver版本号：v2.23.2
入口文件 run_server.py
run_server中通过subprocess.call，用python3运行了同级目录下jms，并传入参数 start all，进入jms.py
首先配置BASE_DIR和APP_DIR全局变量，设置DJANGO_SETTINGS_MODULE值为jumpserver.settings后调用django.steup()，这里就不说了，djang.steup()源码流程可看这里，然后配置logging格式以及必要导入成功检测，并初始化logging对象和创建static和media文件夹，准备工作第一步完成。源码如下：
BASE_DIR = os.path.dirname(os.path.abspath(__file__)) APP_DIR = os.path.join(BASE_DIR, 'apps') os.chdir(APP_DIR) sys.path.insert(0, APP_DIR) os.environ.setdefault("DJANGO_SETTINGS_MODULE", "jumpserver.settings") django.setup() logging.basicConfig(level=logging.DEBUG, format="%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S") try: from jumpserver import const __version__ = const.VERSION except ImportError as e: print("Not found __version__: {}".format(e)) print("Python is: ") logging.info(sys.executable) __version__ = 'Unknown' sys.exit(1) try: from jumpserver.const import CONFIG from common.utils.file import download_file except ImportError as e: print("Import error: {}".format(e)) print("Could not find config file, `cp config_example.yml config.yml`") sys.exit(1) os.environ["PYTHONIOENCODING"] = "UTF-8" logging.basicConfig( format='%(asctime)s %(message)s', level=logging.INFO, datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger() try: os.makedirs(os.path.join(BASE_DIR, "data", "static")) os.makedirs(os.path.join(BASE_DIR, "data", "media")) except: pass
然后到文件最后，创建parse，并准备参数配置，格式化参数生产参数对象args（注：这里的args为全局变量，所以jms中的函数可以直接使用该变量，不用当参数传递），args.action为"start"，因此走start_services，该部分源码如下：
if __name__ == '__main__': parser = argparse.ArgumentParser( description=""" Jumpserver service control tools; Example: \r\n %(prog)s start all -d; """ ) parser.add_argument( 'action', type=str, choices=("start", "stop", "restart", "status", "upgrade_db", "collect_static"), help="Action to run" ) parser.add_argument( "services", type=str, default='all', nargs="*", choices=("all", "web", "task"), help="The service to start", ) parser.add_argument('-d', '--daemon', nargs="?", const=True) parser.add_argument('-w', '--worker', type=int, nargs="?", default=4) parser.add_argument('-f', '--force', nargs="?", const=True) args = parser.parse_args() action = args.action if action == "upgrade_db": upgrade_db() elif action == "collect_static": collect_static() else: start_services()
start_services源码如下：
def start_services(): services = args.services if isinstance(args.services, list) else [args.services] if action == 'start' and {'all', 'web'} & set(services): prepare() start_args = [] if args.daemon: start_args.append('--daemon') if args.worker: start_args.extend(['--worker', str(args.worker)]) if args.force: start_args.append('--force') try: management.call_command(action, *services, *start_args) except KeyboardInterrupt: logging.info('Cancel ...') time.sleep(2) except Exception as exc: logging.error("Start service error {}: {}".format(services, exc)) time.sleep(2)
services为"all"，因此会执行prepare()函数，该函数中分别执行：1.调用django接口检查数据库连接是否正常 2.收集静态文件和数据库迁移 3.清除缓存 4.下载两个ip库，源码较多也简单，就不贴了。最后通过django接口management.call_command调用start并传入all、--daemon、--worker 4、--force参数，这里讲下management.call_command，源码精简贴下，不重要的就不贴了，如下：
def call_command(command_name, *args, **options): """ Call the given command, with the given options and args/kwargs. This is the primary API you should use for calling specific commands. `command_name` may be a string or a command object. Using a string is preferred unless the command object is required for further processing or testing. Some examples: call_command('migrate') call_command('shell', plain=True) call_command('sqlmigrate', 'myapp') from django.core.management.commands import flush cmd = flush.Command() call_command(cmd, verbosity=0, interactive=False) # Do something with cmd ... """ if isinstance(command_name, BaseCommand): # Command object passed in. command = command_name command_name = command.__class__.__module__.split('.')[-1] else: # Load the command object by name. try: app_name = get_commands()[command_name] except KeyError: raise CommandError("Unknown command: %r" % command_name) if isinstance(app_name, BaseCommand): # If the command is already loaded, use it directly. command = app_name else: command = load_command_class(app_name, command_name) # 中间省略 。。。 return command.execute(*args, **defaults)
首先command_name为"start"字符串，走else，再看app_name = get_commands()[command_name]，get_commands()在django.steup()同篇文章django runserver源码中有讲，其实质做的是以字典方式保存django/core/management/commands和所有子app目录/management/commands目录下所有以非"_"开头模块文件名称和app字符串的对应关系，在项目目录下apps/common/management/commands下有start.py文件，因此此时app_name为“common"，最后走load_command_class(app_name, command_name)，load_command_class源码如下：
def load_command_class(app_name, name): """ Given a command name and an application name, return the Command class instance. Allow all errors raised by the import process (ImportError, AttributeError) to propagate. """ module = import_module('%s.management.commands.%s' % (app_name, name)) return module.Command()
回到call_command函数，此时command为start.py中的Command类（一下简称Command类）的实力对象，最后调用对象的execute()方法，Command类继承BaseActionCommand类，BaseActionCommand类未定义execute方法，于是到BaseCommand类的execute方法，不重要的也不贴了，源码如下：
def execute(self, *args, **options): """ Try to execute this command, performing system checks if needed (as controlled by the ``requires_system_checks`` attribute, except if force-skipped). """ # 前面省略。。。 output = self.handle(*args, **options) if output: if self.output_transaction: connection = connections[options.get('database', DEFAULT_DB_ALIAS)] output = '%s\n%s\n%s' % ( self.style.SQL_KEYWORD(connection.ops.start_transaction_sql()), output, self.style.SQL_KEYWORD(connection.ops.end_transaction_sql()), ) self.stdout.write(output) return output
最终调用self.handle()方法，Command类下未定义该方法，于是到BaseActionCommand类，该类handle方法源码如下：
def handle(self, *args, **options): self.initial_util(*args, **options) assert self.action in Action.values, f'The action {self.action} is not in the optional list' _handle = getattr(self, f'_handle_{self.action}', lambda: None) _handle()
initial_util方法中创建ServicesUtil类的实例属性，并赋值为self.util。创建ServicesUtil类时传入参数分别为：run_daemon：True，stop_daemon：False，force_stop：True。services流程讲一下，源码先贴一下：
@classmethod def get_service_objects(cls, service_names, **kwargs): services = set() for name in service_names: method_name = f'{name}_services' if hasattr(cls, method_name): _services = getattr(cls, method_name)() elif hasattr(cls, name): _services = [getattr(cls, name)] else: continue services.update(set(_services)) service_objects = [] for s in services: service_class = cls.get_service_object_class(s.value) if not service_class: continue kwargs.update({ 'name': s.value }) service_object = service_class(**kwargs) service_objects.append(service_object) return service_objects
传入的service_names为["all“]，因此_services为[cls.gunicorn, cls.daphne, cls.flower, cls.celery_ansible, cls.celery_default, cls.beat]，函数内的services为{cls.gunicorn, cls.daphne, cls.flower, cls.celery_ansible, cls.celery_default, cls.beat}，service_objects（返回的services）为services.GunicornService、services.DaphneService、services.FlowerService、services.CeleryDefaultService、services.CeleryAnsibleService、services.BeatService这六个类的实例对象组成的列表。回到handle方法，_handle为self._handle_start()，最后调用该方法，该方法内调用self.util.start_and_watch()方法执行，于是到ServicesUtil.start_and_watch方法，源码如下：
def start_and_watch(self): logging.info(time.ctime()) logging.info(f'JumpServer version {__version__}, more see https://www.jumpserver.org') self.start() if self.run_daemon: self.show_status() with self.daemon_context: self.watch() else: self.watch()
先看self.start()，遍历self._services，也就是遍历上述6个类的实例对象，并调用其start方法，这6个类都继承自BaseService类（jumpserver团队自己定义的服务接口类），且6各类均未定义start方法，于是来到BaseService类的start方法，源码如下：
def start(self): if self.is_running: self.show_status() return self.remove_pid() self.open_subprocess() self.write_pid() self.start_other()
self.is_running是通过对os.kill(self.pid, 0)是否异常来判断该进程号是否在运行，最开始肯定是未运行状态，于是走self.remove_pid()，该方法是通过os.unlink删除tmp文件夹下对应服务的pid文件，然后走self.open_subprocess()，通过subprocess.Popen调用self.cmd，并传入self.cwd和输出文件配置，这里就是真正的开启每个服务了，具体如何开启可以看六个类的cmd属性（property）。然后通过self.write_pid写pid文件，self.start_other是留的一个空函数。回到ServicesUtil.start_and_watch方法，首先self.run_daemon为True，因此调用self.show_status()打印六个服务每个服务是running还是stoped，最后在守护进程上下文内执行self.watch()，self.watch中每隔30秒执行一次self._watch()，self._watch()中是执行六个服务类实例对象的watch方法，服务类对象的watch方法中检查服务是否在运行，如果是未运行状态则重新运行，如果self.watch中捕获到KeyboardInterrupt（键盘退出）异常，则调用self.clean_up()进行清理操作，退出程序。贴下ServicesUtil类的watch()方法，如下：
# -- watch -- def watch(self): while not self.EXIT_EVENT.is_set(): try: _exit = self._watch() if _exit: break time.sleep(self.check_interval) except KeyboardInterrupt: print('Start stop services') break self.clean_up()
Comments