ambari install逻辑详解[二]

# 3.2.2 `execute()` 解读

def execute(self):
    """
    Sets up logging;
    Parses command parameters and executes method relevant to command type
    """

    ##### 3.2.2.1 解析命令行参数 #####
    parser = OptionParser()
    parser.add_option("-o", "--out-files-logging", dest="log_out_files", action="store_true",
                      help="use this option to enable outputting *.out files of the service pre-start")
    (self.options, args) = parser.parse_args()

    self.log_out_files = self.options.log_out_files

    # parse arguments
    if len(args) < 6:
        print "Script expects at least 6 arguments"
        print USAGE.format(os.path.basename(sys.argv[0])) # print to stdout
        sys.exit(1)

    self.command_name = str.lower(sys.argv[1])
    self.command_data_file = sys.argv[2]
    self.basedir = sys.argv[3]
    self.stroutfile = sys.argv[4]
    self.load_structured_out()
    self.logging_level = sys.argv[5]
    Script.tmp_dir = sys.argv[6]
    # optional script arguments for forcing https protocol and ca_certs file
    if len(sys.argv) >= 8:
        Script.force_https_protocol = sys.argv[7]
    if len(sys.argv) >= 9:
        Script.ca_cert_file_path = sys.argv[8]

    logging_level_str = logging._levelNames[self.logging_level]
    Logger.initialize_logger(__name__, logging_level=logging_level_str)

    ##### 3.2.2.2 Windows 环境变量处理 #####
    if OSCheck.is_windows_family():
        reload_windows_env()

    ##### 3.2.2.3 重置 status 命令的结构化输出 #####
    if self.command_name == "status":
        Script.structuredOut = {}
        self.put_structured_out({})

    ##### 3.2.2.4 SSL 设置 #####
    ensure_ssl_using_protocol(Script.get_force_https_protocol_name(), Script.get_ca_cert_file_path())

    ##### 3.2.2.5 加载 JSON 配置文件 #####
    try:
        with open(self.command_data_file) as f:
            Script.config = ConfigDictionary(json.load(f))
            Script.execution_command = ExecutionCommand(Script.config)
            Script.module_configs = Script.execution_command.get_module_configs()
            Script.cluster_settings = Script.execution_command.get_cluster_settings()
            Script.stack_settings = Script.execution_command.get_stack_settings()
            # load passwords here (used on windows to impersonate different users)
            Script.passwords = {}
            for k, v in _PASSWORD_MAP.iteritems():
                if get_path_from_configuration(k, Script.config) and get_path_from_configuration(v, Script.config):
                    Script.passwords[get_path_from_configuration(k, Script.config)] = get_path_from_configuration(v, Script.config)

    except IOError:
        Logger.logger.exception("Can not read json file with command parameters: ")
        sys.exit(1)

    Script.repository_util = RepositoryUtil(Script.config)

    ##### 3.2.2.6 动态选择和执行方法 #####
    try:
        method = self.choose_method_to_execute(self.command_name)
        with Environment(self.basedir, tmp_dir=Script.tmp_dir) as env:
            env.config.download_path = Script.tmp_dir

            if not self.is_hook():
                self.execute_prefix_function(self.command_name, 'pre', env)

            method(env)

            if not self.is_hook():
                self.execute_prefix_function(self.command_name, 'post', env)

    ##### 3.2.2.7 异常处理 #####
    except (ComponentIsNotRunning, ClientComponentHasNoStatus), e:
        traceback.print_exc()
        sys.exit(1)
    except Fail as ex:
        ex.pre_raise()
        raise
    finally:
        ##### 3.2.2.8 版本信息保存 #####
        try:
            if self.should_expose_component_version(self.command_name):
                self.save_component_version_to_structured_out(self.command_name)
        except:
            Logger.exception("Reporting component version failed")

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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

# 3.2.2.1 解析命令行参数

parser = OptionParser()
parser.add_option("-o", "--out-files-logging", dest="log_out_files", action="store_true",
                  help="use this option to enable outputting *.out files of the service pre-start")
(self.options, args) = parser.parse_args()

self.log_out_files = self.options.log_out_files

# parse arguments
if len(args) < 6:
    print "Script expects at least 6 arguments"
    print USAGE.format(os.path.basename(sys.argv[0])) # print to stdout
    sys.exit(1)

self.command_name = str.lower(sys.argv[1])
self.command_data_file = sys.argv[2]
self.basedir = sys.argv[3]
self.stroutfile = sys.argv[4]
self.load_structured_out()
self.logging_level = sys.argv[5]
Script.tmp_dir = sys.argv[6]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

提示

作用： 解析和校验脚本的命令行参数，提取关键路径、日志文件、命令类型等信息，为后续步骤打基础。参数不足会直接退出，保障脚本健壮性和可追溯性。

# 3.2.2.2 Windows 环境变量处理

if OSCheck.is_windows_family():
    reload_windows_env()

1
2

笔记

仅 Windows 平台执行，重新加载环境变量，防止脚本运行时环境缺失，增强跨平台兼容性。

# 3.2.2.3 重置 status 命令的结构化输出

if self.command_name == "status":
    Script.structuredOut = {}
    self.put_structured_out({})

1
2
3

提示

保证每次 status 状态检查的输出是干净的，避免“脏数据”影响运维自动化判断。

# 3.2.2.4 SSL 设置

ensure_ssl_using_protocol(Script.get_force_https_protocol_name(), Script.get_ca_cert_file_path())

笔记

SSL 协议和 CA 证书路径配置，确保所有命令执行都具备合规的安全保障。

# 3.2.2.5 加载 JSON 配置文件

try:
    with open(self.command_data_file) as f:
        Script.config = ConfigDictionary(json.load(f))
        Script.execution_command = ExecutionCommand(Script.config)
        Script.module_configs = Script.execution_command.get_module_configs()
        Script.cluster_settings = Script.execution_command.get_cluster_settings()
        Script.stack_settings = Script.execution_command.get_stack_settings()
        # load passwords here (used on windows to impersonate different users)
        Script.passwords = {}
        for k, v in _PASSWORD_MAP.iteritems():
            if get_path_from_configuration(k, Script.config) and get_path_from_configuration(v, Script.config):
                Script.passwords[get_path_from_configuration(k, Script.config)] = get_path_from_configuration(v, Script.config)

except IOError:
    Logger.logger.exception("Can not read json file with command parameters: ")
    sys.exit(1)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

提示

加载组件具体的执行参数、模块配置、集群配置、Stack 配置等，为后续 install、start、stop 等方法调用做上下文准备。也兼容了 Windows 下切换用户需要的密码管理。

# 3.2.2.6 动态选择和执行方法

try:
    method = self.choose_method_to_execute(self.command_name)
    with Environment(self.basedir, tmp_dir=Script.tmp_dir) as env:
        env.config.download_path = Script.tmp_dir

        if not self.is_hook():
            self.execute_prefix_function(self.command_name, 'pre', env)

        method(env)

        if not self.is_hook():
            self.execute_prefix_function(self.command_name, 'post', env)

1
2
3
4
5
6
7
8
9
10
11
12

方法选择与调用机制如下：

def choose_method_to_execute(self, command_name):
    """
    Returns a callable object that should be executed for a given command.
    """
    self_methods = dir(self)
    Logger.info("=========== Methods and attributes in this class: {0}".format(dir(self)))
    if not command_name in self_methods:
        raise Fail("Script '{0}' has no method '{1}'".format(sys.argv[0], command_name))
    method = getattr(self, command_name)
    return method

1
2
3
4
5
6
7
8
9
10

def execute_prefix_function(self, command_name, afix, env):
    """
    Execute action afix (prefix or suffix) based on command_name and afix type
    example: command_name=start, afix=pre will result in execution of self.pre_start(env) if exists
    """
    self_methods = dir(self)
    method_name = "{0}_{1}".format(afix, command_name)
    if not method_name in self_methods:
        Logger.logger.debug("Action afix '{0}' not present".format(method_name))
        return
    Logger.logger.debug("Execute action afix: {0}".format(method_name))
    method = getattr(self, method_name)
    method(env)

1
2
3
4
5
6
7
8
9
10
11
12
13

提示

先通过方法名反射，动态查找（如 install/start/stop）并执行。若有 pre_xxx/post_xxx 钩子，会自动在主流程前后调用（如 pre_install, post_install），极大增强了可扩展性。无需硬编码流程，所有组件都可通过同一机制进行前后扩展。

# 3.2.2.7 异常处理

except (ComponentIsNotRunning, ClientComponentHasNoStatus), e:
    traceback.print_exc()
    sys.exit(1)
except Fail as ex:
    ex.pre_raise()
    raise

1
2
3
4
5
6

注意

精准捕获并记录组件未运行或无状态等异常。保证自动化脚本出错时，日志清晰、退出明确，利于自动化调度平台追溯。

# 3.2.2.8 版本信息保存

finally:
    try:
        if self.should_expose_component_version(self.command_name):
            self.save_component_version_to_structured_out(self.command_name)
    except:
        Logger.exception("Reporting component version failed")