jumpserver运行源码

本文所使用的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()


  • 作者:合十
  • 发表时间:2022年6月26日 16:33
  • 更新时间:2022年8月18日 22:13
  • 所属分类:我用Python

Comments

该文章还未收到评论,点击下方评论框开始评论吧~