saltstack源码解析之master启动流程

阅读数：108次

ENV: Version: 2016.11.0, Transport: tcp, Multi-Master

1. 类图和重要方法说明

salt-master启动流程

2. 源码解析

2.1 初始化（salt.master.Master.__init__）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# salt/master.py
class Master(SMaster):
    '''
    The salt master server
    '''
    def __init__(self, opts):
        '''
        Create a salt master server instance

        :param dict: The salt options
        '''
        if HAS_ZMQ:
            # Warn if ZMQ < 3.2
            try:
                zmq_version_info = zmq.zmq_version_info()
            except AttributeError:
                # PyZMQ <= 2.1.9 does not have zmq_version_info, fall back to
                # using zmq.zmq_version() and build a version info tuple.
                zmq_version_info = tuple(
                    [int(x) for x in zmq.zmq_version().split('.')]
                )
            if zmq_version_info < (3, 2):
                log.warning(
                    'You have a version of ZMQ less than ZMQ 3.2! There are '
                    'known connection keep-alive issues with ZMQ < 3.2 which '
                    'may result in loss of contact with minions. Please '
                    'upgrade your ZMQ!'
                )
        SMaster.__init__(self, opts)
  • 检查zmq版本
  • 调用SMaster.init(self, opts)，进行opts、master_key和key的初始化
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    # salt/master.py
    class SMaster(object):
        '''
        Create a simple salt-master, this will generate the top-level master
        '''
        secrets = {}  # mapping of key -> {'secret': multiprocessing type, 'reload': FUNCTION}

        def __init__(self, opts):
            '''
            Create a salt master server instance

            :param dict opts: The salt options dictionary
            '''
            self.opts = opts
            self.master_key = salt.crypt.MasterKeys(self.opts)
            self.key = self.__prep_key()

2.2 启动（salt.master.Master.start）

1.启动前检查，Master._pre_flight

  • os.chdir(‘/‘)
  • 根据fileserver_backend配置，创建Fileserver。这里用到了LazyLoader，先创建空的module，在实际调用的时候才进行加载，而不是在启动时全部import
    1
    2
    3
    4
    5
    6
    7
    # salt/loader.py
    class LazyLoader(salt.utils.lazy.LazyDict):
        '''
        A pseduo-dictionary which has a set of keys which are the
        name of the module and function, delimited by a dot. When
        the value of the key is accessed, the function is then loaded
        from disk and into memory.
1
2
3
4
5
6
7
8
9
# salt/loader.py
def _generate_module(name):
    if name in sys.modules:
        return

    code = "'''Salt loaded {0} parent module'''".format(name.split('.')[-1])
    module = imp.new_module(name)
    exec(code, module.__dict__)
    sys.modules[name] = module
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def __getattr__(self, mod_name):
    '''
    Allow for "direct" attribute access-- this allows jinja templates to
    access things like `salt.test.ping()`
    '''
    # if we have an attribute named that, lets return it.
    try:
        return object.__getattr__(self, mod_name)  # pylint: disable=no-member
    except AttributeError:
        pass

    # otherwise we assume its jinja template access
    if mod_name not in self.loaded_modules and not self.loaded:
        for name in self._iter_files(mod_name):
            if name in self.loaded_files:
                continue
            # if we got what we wanted, we are done
            if self._load_module(name) and mod_name in self.loaded_modules:
                break
    if mod_name in self.loaded_modules:
        return self.loaded_modules[mod_name]
    else:
        raise AttributeError(mod_name)
  • 创建pillar_cache文件夹
  • 检查git_pillar

    This external pillar allows for a Pillar top file and Pillar SLS files to be sourced from a git repository.

1
2
3
ext_pillar:
  - git: master https://gitserver/git-pillar.git
  - git: dev https://gitserver/git-pillar.git

2.注册信号处理程序

  • enable_sigusr1_handler。打印stack trace

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    # salt/utils/debug.py
    def _handle_sigusr1(sig, stack):
        '''
        Signal handler for SIGUSR1, only available on Unix-like systems
        '''
        # When running in the foreground, do the right  thing
        # and spit out the debug info straight to the console
        if sys.stderr.isatty():
            output = sys.stderr
            _makepretty(output, stack)
        else:
            filename = 'salt-debug-{0}.log'.format(int(time.time()))
            destfile = os.path.join(tempfile.gettempdir(), filename)
            with salt.utils.fopen(destfile, 'w') as output:
                _makepretty(output, stack)

  • enable_sigusr2_handler。YAPPI开关

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    # salt/utils/debug.py
    def _handle_sigusr2(sig, stack):
        '''
        Signal handler for SIGUSR2, only available on Unix-like systems
        '''
        try:
            import yappi
        except ImportError:
            return
        if yappi.is_running():
            yappi.stop()
            filename = 'callgrind.salt-{0}-{1}'.format(int(time.time()), os.getpid())
            destfile = os.path.join(tempfile.gettempdir(), filename)
            yappi.get_func_stats().save(destfile, type='CALLGRIND')
            if sys.stderr.isatty():
                sys.stderr.write('Saved profiling data to: {0}\n'.format(destfile))
            yappi.clear_stats()
        else:
            if sys.stderr.isatty():
                sys.stderr.write('Profiling started\n')
            yappi.start()

3.设置max open files，Master.__set_max_open_files
4.初始化SMaster.secrets[‘aes’]，在进行消息的发送时加密使用
5.创建ProcessManager，后续创建的各组件进程都会添加进来

1
2
3
4
# salt/master.py
            log.info('Creating master process manager')
            # Since there are children having their own ProcessManager we should wait for kill more time.
            self.process_manager = salt.utils.process.ProcessManager(wait_for_kill=5)

6.创建publisher process。用于分发消息给minion

1
2
3
4
5
6
7
# salt/master.py
            pub_channels = []
            log.info('Creating master publisher process')
            for transport, opts in iter_transport_opts(self.opts):
                chan = salt.transport.server.PubServerChannel.factory(opts)
                chan.pre_fork(self.process_manager)
                pub_channels.append(chan)

7.创建event publisher process。Master内部的消息总线，进行信息交换

1
2
3
# salt/master.py
            log.info('Creating master event publisher process')
            self.process_manager.add_process(salt.utils.event.EventPublisher, args=(self.opts,))

8.创建reactor。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# salt/master.py
            if self.opts.get('reactor'):
                if isinstance(self.opts['engines'], list):
                    rine = False
                    for item in self.opts['engines']:
                        if 'reactor' in item:
                            rine = True
                            break
                    if not rine:
                        self.opts['engines'].append({'reactor': {}})
                else:
                    if 'reactor' not in self.opts['engines']:
                        log.info('Enabling the reactor engine')
                        self.opts['engines']['reactor'] = {}

            salt.engines.start_engines(self.opts, self.process_manager)

Salt’s Reactor system gives Salt the ability to trigger actions in response to an event. It is a simple interface to watching Salt’s event bus for event tags that match a given pattern and then running one or more commands in response.
This system binds sls files to event tags on the master. These sls files then define reactions. This means that the reactor system has two parts. First, the reactor option needs to be set in the master configuration file. The reactor option allows for event tags to be associated with sls reaction files. Second, these reaction files use highdata (like the state system) to define reactions to be executed.

1
2
3
4
5
6
7
8
9
engines:
  - reactor:
      refresh_interval: 60
      worker_threads: 10
      worker_hwm: 10000

reactor:
  - 'salt/cloud/*/destroyed':
    - /srv/reactor/destroy/*.sls

9.创建maintenance process。A generalized maintenance process which performs maintenance routines.

1
2
3
4
# salt/master.py
            # must be after channels
            log.info('Creating master maintenance process')
            self.process_manager.add_process(Maintenance, args=(self.opts,))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# salt/master.py
    def run(self):
        '''
        This is the general passive maintenance process controller for the Salt
        master.

        This is where any data that needs to be cleanly maintained from the
        master is maintained.
        '''
        salt.utils.appendproctitle('Maintenance')

        # init things that need to be done after the process is forked
        self._post_fork_init()

        # Make Start Times
        last = int(time.time())
        # Clean out the fileserver backend cache
        salt.daemons.masterapi.clean_fsbackend(self.opts)
        # Clean out pub auth
        salt.daemons.masterapi.clean_pub_auth(self.opts)

        old_present = set()
        while True:
            now = int(time.time())
            if (now - last) >= self.loop_interval:
                salt.daemons.masterapi.clean_old_jobs(self.opts)
                salt.daemons.masterapi.clean_expired_tokens(self.opts)
            self.handle_search(now, last)
            self.handle_git_pillar()
            self.handle_schedule()
            self.handle_key_cache()
            self.handle_presence(old_present)
            self.handle_key_rotate(now)
            salt.daemons.masterapi.fileserver_update(self.fileserver)
            salt.utils.verify.check_max_open_files(self.opts)
            last = now
            time.sleep(self.loop_interval)

10.创建event return process。

1
2
3
4
# salt/master.py
            if self.opts.get('event_return'):
                log.info('Creating master event return process')
                self.process_manager.add_process(salt.utils.event.EventReturn, args=(self.opts,))

All Salt commands will return the command data back to the master. Specifying returners will ensure that the data is also sent to the specified returner interfaces.

11.创建ext_processes process

1
2
3
4
5
6
7
8
9
10
11
12
13
# salt/master.py
            ext_procs = self.opts.get('ext_processes', [])
            for proc in ext_procs:
                log.info('Creating ext_processes process: {0}'.format(proc))
                try:
                    mod = '.'.join(proc.split('.')[:-1])
                    cls = proc.split('.')[-1]
                    _tmp = __import__(mod, globals(), locals(), [cls], -1)
                    cls = _tmp.__getattribute__(cls)
                    self.process_manager.add_process(cls, args=(self.opts,))
                except Exception:
                    log.error(('Error creating ext_processes '
                            'process: {0}').format(proc))

In addition to the processes that the Salt master automatically spawns, it is possible to configure it to start additional custom processes.
This is useful if a dedicated process is needed that should run throughout the life of the Salt master. For periodic independent tasks, a scheduled runner may be more appropriate.
Processes started in this way will be restarted if they die and will be killed when the Salt master is shut down.

1
2
3
ext_processes:
  - mymodule.TestProcess
  - mymodule.AnotherProcess

12.创建halite process。DEPRECATED: A client-side web application interface to a running Salt infrastructure

1
2
3
4
# salt/master.py
            if HAS_HALITE and 'halite' in self.opts:
                log.info('Creating master halite process')
                self.process_manager.add_process(Halite, args=(self.opts['halite'],))

13.创建concache process

1
2
3
4
5
6
7
8
# salt/master.py
            # TODO: remove, or at least push into the transport stuff (pre-fork probably makes sense there)
            if self.opts['con_cache']:
                log.info('Creating master concache process')
                self.process_manager.add_process(ConnectedCache, args=(self.opts,))
                # workaround for issue #16315, race condition
                log.debug('Sleeping for two seconds to let concache rest')
                time.sleep(2)

con_cache
Default: False
If max_minions is used in large installations, the master might experience high-load situations because of having to check the number of connected minions for every authentication. This cache provides the minion-ids of all connected minions to all MWorker-processes and greatly improves the performance of max_minions.

14.创建request server process。处理salt命令和minion处理结果

1
2
3
4
5
6
7
8
9
10
11
12
# salt/master.py
            log.info('Creating master request server process')
            kwargs = {}
            if salt.utils.is_windows():
                kwargs['log_queue'] = salt.log.setup.get_multiprocessing_logging_queue()
                kwargs['secrets'] = SMaster.secrets

            self.process_manager.add_process(
                ReqServer,
                args=(self.opts, self.key, self.master_key),
                kwargs=kwargs,
                name='ReqServer')

创建ReqServerChannel和MWorker

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

微信分享二维码

Related Issues not found

Please contact @winway to initialize the comment

本站总访问量66283次 | 本站访客数51309人
Fork me on GitHub