salt 源码之执行流程

1.salt 命令的开始脚本 [root@zhu1 ~]# which salt /usr/bin/salt [root@zhu1 ~]# cat /usr/bin/salt #!/usr/bin/python2.6 ''' Publish commands to the salt system from the command line on the master. ''' from salt.scripts import salt_main if __name__ == '__main__': salt_main() 可以看到salt命令实际上是调用的是： salt.scripts文件中的salt_main()函数而已

client = salt.cli.salt.SaltCMD() client.run() salt_main函数建立一个salt.cli.salt.SaltCMD实例client，然后执行该实例的run方法 run方法执行内容如下： self.parse_args() #解析命令行选项 local = salt.client.get_local_client(self.get_config_file_path(), skip_perm_errors=skip_perm_errors) 生成一个字典kwargs ；该字典稍后会作为参数传递给执行函数以 salt '*' test.ping 为例 kwargs = { 'tgt': '*', 'show_timeout': True, 'delimiter': ':', 'expr_form': 'glob', 'timeout': 5, 'arg': [], 'fun': 'test.ping', 'show_jid': False } 执行： local.cmd_cli(**kwargs) 在看cmd_cli方法执行内容前，需看local实例具体做的那些事情

salt 执行命令工作流程(以 salt '*' test.ping 为例): 1.调用self.parse_args()方法解析命令行参数 2.实例化 local = salt.client.get_local_client('/etc/salt/master',False) salt.client.get_local_client 是一个函数,该函数返回LocalClient()实例 3.实例化 LocalClient() 4.初始化 self.event = salt.utils.event.get_event('master',/var/run/salt/master,zeromq,'self.opts','listen=False') 5.salt.utils.event.get_event()函数返回salt.utils.event.SaltEvent()类 self.puburi = /var/run/salt/master/master_event_pub.ipc self.pulluri = /var/run/salt/master/master_event_pull.ipc 初始化 self.utils = salt.loader.utils(self.opts) 初始化 self.functions = salt.loader.minion_mods(self.opts, utils=self.utils) 初始化 self.returners = salt.loader.returners(self.opts, self.functions) 6.生成字典 kwargs = {'tgt': 'zhu1', 'show_timeout': True, 'delimiter': ':', 'expr_form': 'glob', 'timeout': 5, 'arg': [], 'fun': 'test.ping', 'show_jid': False} 7.开始执行: local.cmd_cli(**kwargs)方法 8.在local.cmd_cli方法中调用self.run_job方法 9.在local.run_job方法中调用self.pub方法 10.在local.pub方法中调用self._prep_pub方法 11.self._prep_pub方法返回一个字典,该字典内容为发送给salt-master的参数格式: payload_kwargs = {'tgt_type': 'glob', 'jid': '', 'key': 'rtXXpQ75m9tp0RgzNxYJuhX0ulKOHW4qKO3cMerv++4a3HDCTArXW+F1UNbZAqmc9MXAdHbHl/U=', 'tgt': 'zhu1', 'arg': [], 'fun': 'test.ping', 'kwargs': {'delimiter': ':', 'show_timeout': True, 'show_jid': False}, 'cmd': 'publish', 'ret': '', 'user': 'root'} 12.salt发送的master的地址为 master_uri = tcp://127.0.0.1:4506 13.实例化channel = salt.transport.Channel.factory(self.opts,crypt='clear',master_uri=tcp://127.0.0.1:4506) salt.transport.Channel.factory 函数返回salt.transport.client.ReqChannel.factory() salt.transport.client.ReqChannel.factory()返回SyncWrapper(AsyncReqChannel.factory, (opts,), kwargs) AsyncReqChannel.factory返回 salt.transport.zeromq.AsyncZeroMQReqChannel(opts, **kwargs) AsyncZeroMQReqChannel继承于salt.transport.client.ReqChannel AsyncZeroMQReqChannel在创建类时会执行self.message_client = AsyncReqMessageClient(self.opts,self.master_uri,io_loop=self._io_loop) 在AsyncReqMessageClient类实例化时执行如下部分代码： self.context = zmq.Context() self.socket = self.context.socket(zmq.REQ) self.socket.connect(self.addr) 14.执行self.event.connect_pub(timeout=timeout) 15.在connect_pub中执行self.subscriber = salt.transport.ipc.IPCMessageSubscriber(/var/run/salt/master/master_event_pub.ipc) 16.IPCMessageSubscriber连接------------->/var/run/salt/master/master_event_pub.ipc 17.执行payload = channel.send(payload_kwargs, timeout=timeout) 执行发送: channel.send()--->AsyncZeroMQReqChannel.send() 执行发送: self.message_client.send() 执行发送 AsyncReqMessageClient.send() 18.发送tcp://127.0.0.1:4506后返回的payload = {'load': {'jid': '20161227141413152668', 'minions': ['zhu1']}, 'enc': 'clear'} 19.在local.run_job函数中执行self.pub返回的数据: pub_data ={'jid': '20161227141413152668', 'minions': ['zhu1']} 20.在local.run_job函数中执行self._check_pub_data(pub_data)检查返回的数据 opts.get('order_masters') = False 21.开始执行 self.event.subscribe('salt/job/20161227141413152668')根据jid订阅匹配的事件返回pub_data = {'jid': '20161227141413152668', 'minions': ['zhu1']} 22.在cmd_cli函数中 self.run_job方法返回数据 pub_data = {'jid': '20161227141413152668', 'minions': ['zhu1']} 23.在cmd_cli方法中 self.get_cli_event_returns一定会执行 24.开始执行self.get_cli_event_returns(20161227141413152668,['zhu1'],5,zhu1,glob,False,False) 25.开始执行: get_iter_returns(20161227141413152668,['zhu1']) 26.开始执行get_iter_returns(20161227141413152668,['zhu1'])方法;一直监听事件系统直到所等待的job数据到来 27.执行returners = self.returners['local_cache.get_load'] <class 'salt.loader.LazyLoader'> jid_dir=/var/cache/salt/master/jobs/bc/21bfd15870cdc50f82a8555eba79e7 load_fn=/var/cache/salt/master/jobs/bc/21bfd15870cdc50f82a8555eba79e7/.load.p minions_cache = ['/var/cache/salt/master/jobs/bc/21bfd15870cdc50f82a8555eba79e7/.minions.p'] 28. 开始执行 self.get_returns_no_block('salt/job/20161227141413152668') ret = {'tag': '20161227141413152668', 'data': {'_stamp': '2016-12-27T06:14:13.156325', 'minions': ['zhu1']}} ret = {'tag': 'salt/job/20161227141413152668/new', 'data': {'tgt_type': 'glob', 'jid': '20161227141413152668', 'tgt': 'zhu1', '_stamp': '2016-12-27T06:14:13.167890', 'user': 'root', 'arg': [], 'fun': 'test.ping', 'minions': ['zhu1']}} ret = {'tag': 'salt/job/20161227141413152668/ret/zhu1', 'data': {'fun_args': [], 'jid': '20161227141413152668', 'return': True, 'retcode': 0, 'success': True, 'cmd': '_return', '_stamp': '2016-12-27T06:14:13.294696', 'fun': 'test.ping', 'id': 'zhu1'}} fn_ret={'zhu1': {'retcode': 0, 'ret': True}} full_ret={'zhu1': {'retcode': 0, 'ret': True}} zhu1: True 开始执行: self._clean_up_subscriptions(20161227141413152668)

salt 源码之 执行流程

相关阅读