AMBARI-23508: Update execution_command module to handle data type con… (#957)

* AMBARI-23508: Update execution_command module to handle data type conversion issue

* AMBARI-23508: Update execution_command module to handle data type conversion issue

ambari-common/src/main/python/resource_management/libraries/execution_command/execution_command.py

@@ -27,26 +27,31 @@ from resource_management.libraries.execution_command import module_configs
 
 class ExecutionCommand(object):
   """
-  The class maps to a command.json. All command related info should be retrieved from this class object
+  The class has two private objects: _execution_command maps to a command.json and
+  _module_config maps to configurations block and configuratonAttributes within this
+   command.json. This class provides a set of APIs to retrieve all command related info.
+   The caller should not access to command.json to get the configuration info directly.
   """
 
   def __init__(self, command):
     """
-    _execution_command is an internal dict object maps to command.json
+    _execution_command is a wrapper object of a dict object maps to command.json
+    _module_config is an wrapper object of configuration and configurationAttributes dict
+    objects maps to configuration and configurationAttributes blocks within command.json
     :param command: json string or a python dict object
     """
-    self._execution_command = command
-    self._module_configs = module_configs.ModuleConfigs(self.__get_value("configurations"), self.__get_value("configurationAttributes"))
+    self.__execution_command = command
+    self.__module_configs = module_configs.ModuleConfigs(self.__get_value("configurations"), self.__get_value("configurationAttributes"))
 
   def __get_value(self, key, default_value=None):
     """
     A private method to query value with the full path of key, if key does not exist, return default_value
-    :param key:
-    :param default_value:
-    :return:
+    :param key: query key string
+    :param default_value: if key does not exist, return this value
+    :return: the value maps to query key
     """
-    sub_keys = key.split('/')
-    value = self._execution_command
+    sub_keys = filter(None, key.split('/'))
+    value = self.__execution_command
     try:
       for sub_key in sub_keys:
         if not sub_key in value:
@@ -58,7 +63,7 @@ class ExecutionCommand(object):
 
   def get_value(self, query_string, default_value=None):
     """
-    Query config attribute from execution_command directly
+    Generic query API to retrieve config attribute from execution_command directly
     :param query_string: full query key string
     :param default_value: if key does not exist, return default value
     :return: config attribute
@@ -70,31 +75,55 @@ class ExecutionCommand(object):
   """
 
   def get_module_configs(self):
-    return self._module_configs
+    return self.__module_configs
 
   def get_module_name(self):
+    """
+    Retrieve service name from command.json, i.e 'zookeeper', 'hdfs'
+    :return: service name
+    """
     return self.__get_value("serviceName")
 
   def get_component_type(self):
-    return self.__get_value("role", "")
+    """
+    Retrieve host role from command.json, i.e "role": "ZOOKEEPER_SERVER"
+    :return:
+    """
+    return self.__get_value("role")
 
   def get_component_instance_name(self):
     """
     At this time it returns hardcoded 'default' name
-    :return:
+    :return: 'default' string
     """
     return "default"
 
   def get_servicegroup_name(self):
+    """
+    Retrieve servicegroup name from command.json, servicegroup has 1:1 map to mpack at this time
+    :return: servicegroup name
+    """
     return self.__get_value("serviceGroupName")
 
   def get_cluster_name(self):
+    """
+    Retrieve cluster name from command.json
+    :return: cluster name
+    """
     return self.__get_value("clusterName")
 
   def get_repository_file(self):
+    """
+    Retrieve respository dict about mpack info from command.json
+    :return: repository file name
+    """
     return self.__get_value("repositoryFile")
 
   def get_local_components(self):
+    """
+    Retrieve a list of service components from command.json. i.e localComponents": ["ZOOKEEPER_CLIENT"]
+    :return: list of components
+    """
     return self.__get_value("localComponents", [])
 
   """
@@ -102,51 +131,120 @@ class ExecutionCommand(object):
   """
 
   def get_jdk_location(self):
+    """
+    Retrieve URL of jdk from command.json. i.e "jdk_location": "http://c7302.ambari.apache.org:8080/resources/"
+    :return: jdk url string
+    """
     return self.__get_value("ambariLevelParams/jdk_location")
 
   def get_jdk_name(self):
+    """
+    Retrieve jdk name from command.json. i.e "jdk_name": "jdk-8u112-linux-x64.tar.gz"
+    :return: jdk name string
+    """
     return self.__get_value("ambariLevelParams/jdk_name")
 
   def get_java_home(self):
+    """
+    Retrieve java home from command.json. i.e "java_home": "/usr/jdk64/jdk1.8.0_112"
+    :return: java home string
+    """
     return self.__get_value("ambariLevelParams/java_home")
 
   def get_java_version(self):
-    return int(self.__get_value("ambariLevelParams/java_version"))
+    """
+    Retrieve java version from command.json, i.e "java_version": "8", note "8" will convert to 8
+    :return: an integer represents java version
+    """
+    from resource_management.libraries.functions.expect import expect_v2
+    return expect_v2("ambariLevelParams/java_version", int)
 
   def get_jce_name(self):
+    """
+    Retrieve jce name from command.json, i.e "jce_name": "jce_policy-8.zip"
+    :return: jce name string
+    """
     return self.__get_value("ambariLevelParams/jce_name")
 
   def get_db_driver_file_name(self):
+    """
+    Retrieve database driver file name, i.e "db_driver_filename": "mysql-connector-java.jar"
+    :return: DB driver name string
+    """
     return self.__get_value('ambariLevelParams/db_driver_filename')
 
   def get_db_name(self):
+    """
+    Retrieve database name, i.e Ambari server's db name is "db_name": "ambari"
+    :return: DB name
+    """
     return self.__get_value('ambariLevelParams/db_name')
 
   def get_oracle_jdbc_url(self):
+    """
+    Retrieve oracle database jdbc driver url,
+    i.e "oracle_jdbc_url": "http://c7302.ambari.apache.org:8080/resources//ojdbc6.jar"
+    :return: oracle jdbc url
+    """
     return self.__get_value('ambariLevelParams/oracle_jdbc_url')
 
   def get_mysql_jdbc_url(self):
+    """
+    Retrieve mysql database jdbc driver url,
+    i.e "mysql_jdbc_url": "http://c7302.ambari.apache.org:8080/resources//mysql-connector-java.jar"
+    :return: mysql jdbc url
+    """
     return self.__get_value('ambariLevelParams/mysql_jdbc_url')
 
   def get_agent_stack_retry_count(self):
-    return int(self.__get_value('ambariLevelParams/agent_stack_retry_count', 5))
+    """
+    Retrieve retry count for stack deployment on agent,
+    i.e "agent_stack_retry_count": "5"
+    :return: retry count for stack deployment on agent
+    """
+    from resource_management.libraries.functions.expect import expect_v2
+    return expect_v2('ambariLevelParams/agent_stack_retry_count', int, 5)
 
   def check_agent_stack_want_retry_on_unavailability(self):
+    """
+    Check if retry is needed when agent is not reachable
+    :return: True or False
+    """
     return self.__get_value('ambariLevelParams/agent_stack_retry_on_unavailability')
 
   def get_ambari_server_host(self):
+    """
+    Retrieve ambari server host url from command.json, i.e "ambari_server_host": "c7302.ambari.apache.org"
+    :return: ambari server url
+    """
     return self.__get_value("ambariLevelParams/ambari_server_host")
 
   def get_ambari_server_port(self):
+    """
+    Retrieve ambar server port number from command.json, i.e "ambari_server_port": "8080"
+    :return: amabari server port number
+    """
     return self.__get_value("ambariLevelParams/ambari_server_port")
 
   def is_ambari_server_use_ssl(self):
+    """
+    Check if ssl is needed to connect to ambari server
+    :return: True or False
+    """
     return self.__get_value("ambariLevelParams/ambari_server_use_ssl", False)
 
   def is_host_system_prepared(self):
+    """
+    Check a global flag enabling or disabling the sysprep feature
+    :return: True or False
+    """
     return self.__get_value("ambariLevelParams/host_sys_prepped", False)
 
   def is_gpl_license_accepted(self):
+    """
+    Check ambari server accepts gpl license or not
+    :return: True or False
+    """
     return self.__get_value("ambariLevelParams/gpl_license_accepted", False)
 
   """
@@ -154,18 +252,38 @@ class ExecutionCommand(object):
   """
 
   def get_mpack_name(self):
+    """
+    Retrieve mpack name from command.json, i.e "stack_name": "HDPCORE"
+    :return: mpack name string
+    """
     return self.__get_value("clusterLevelParams/stack_name")
 
   def get_mpack_version(self):
+    """
+    Retrieve mpack version from command.json, i.e "stack_version": "1.0.0-b224"
+    :return: mpack version string
+    """
     return self.__get_value("clusterLevelParams/stack_version")
 
   def get_user_groups(self):
+    """
+    Retrieve ambari server user groups, i.e "user_groups": "{\"zookeeper\":[\"hadoop\"],\"ambari-qa\":[\"hadoop\"]}"
+    :return: a user group dict object
+    """
     return self.__get_value("clusterLevelParams/user_groups")
 
   def get_group_list(self):
+    """
+    Retrieve a list of user groups from command.json, i.e "group_list": "[\"hadoop\"]"
+    :return: a list of groups
+    """
     return self.__get_value("clusterLevelParams/group_list")
 
   def get_user_list(self):
+    """
+    Retrieve a list of users from command.json, i.e "user_list": "[\"zookeeper\",\"ambari-qa\"]"
+    :return: a list of users
+    """
     return self.__get_value("clusterLevelParams/user_list")
 
   """
@@ -173,12 +291,25 @@ class ExecutionCommand(object):
   """
 
   def get_host_name(self):
+    """
+    Retrieve host name on which ambari agent is running
+    :return: host name
+    """
     return self.__get_value("agentLevelParams/hostname")
 
   def check_agent_config_execute_in_parallel(self):
+    """
+    Check if config commands can be executed in parallel in ambari agent
+    :return: True or False
+    """
     return int(self.__get_value("agentConfigParams/agent/parallel_execution", 0))
 
   def get_agent_cache_dir(self):
+    """
+    The root directory in which ambari agent stores cache or log etc,
+    i.e "agentCacheDir": "/var/lib/ambari-agent/cache"
+    :return: the cache directory path
+    """
     return self.__get_value('agentLevelParams/agentCacheDir')
 
   """
@@ -196,6 +327,10 @@ class ExecutionCommand(object):
   """
 
   def check_unlimited_key_jce_required(self):
+    """
+    Check unlimited jce key is required or not
+    :return: True or False
+    """
     return self.__get_value('componentLevelParams/unlimited_key_jce_required', False)
 
   """
@@ -210,6 +345,10 @@ class ExecutionCommand(object):
     return self.__get_value("commandParams/version")
 
   def check_command_retry_enabled(self):
+    """
+    Check need to retry command or not
+    :return: True or False
+    """
     return self.__get_value('commandParams/command_retry_enabled', False)
 
   def check_upgrade_direction(self):
@@ -225,10 +364,10 @@ class ExecutionCommand(object):
     return self.__get_value('commandParams/update_files_only', False)
 
   def get_deploy_phase(self):
-    return self.__get_value('commandParams/phase', '')
+    return self.__get_value('commandParams/phase')
 
   def get_dfs_type(self):
-    return self.__get_value('commandParams/dfs_type', '')
+    return self.__get_value('commandParams/dfs_type')
 
   def get_module_package_folder(self):
     return self.__get_value('commandParams/service_package_folder')
@@ -251,6 +390,18 @@ class ExecutionCommand(object):
   def check_only_update_files(self):
     return self.__get_value('commandParams/update_files_only', False)
 
+  def get_desired_namenode_role(self):
+    """
+    The desired role is only available during a Non-Rolling Upgrade in HA.
+    The server calculates which of the two NameNodes will be the active,
+    and the other the standby since they are started using different commands.
+    :return: the node's role (active or standby)
+    """
+    return self.__get_value('commandParams/desired_namenode_role')
+
+  def get_node(self, type):
+    key = "commandParams/" + type + "node"
+    return self.__get_value(key)
 
   """
   Role related variables section
@@ -269,6 +420,10 @@ class ExecutionCommand(object):
       key = "clusterHostInfo/" + component_name
     return self.__get_value(key, [])
 
+  def get_component_host(self, component_name):
+    key = "clusterHostInfo/" + component_name + "_host"
+    return self.__get_value(key, [])
+
   def get_all_hosts(self):
     return self.__get_value('clusterHostInfo/all_hosts', [])
 

ambari-common/src/main/python/resource_management/libraries/execution_command/module_configs.py

@@ -31,6 +31,12 @@ class ModuleConfigs(object):
     self.__module_config_attributes = configAttributes
 
   def get_all_attributes(self, module_name, config_type):
+    """
+    Retrieve attributes from /configurationAttributes/config_type
+    :param module_name:
+    :param config_type:
+    :return:
+    """
     if config_type not in self.__module_config_attributes:
       return {}
     try:
@@ -64,3 +70,5 @@ class ModuleConfigs(object):
       return default
 
 
+
+

ambari-common/src/main/python/resource_management/libraries/functions/expect.py

@@ -62,3 +62,39 @@ def expect(name, expected_type, default_value=None):
     except (ValueError, TypeError):
       raise Fail("Configuration {0} expected to be number, but found '{1}'".format(name, value))
   return value
+
+
+def expect_v2(name, expected_type, default_value=None):
+  """
+  Expect configuration to be of certain type. If it is not, give a reasonable error message to user.
+
+  Optionally if the configuration is not found default_value for it can be returned.
+  """
+
+  value = Script.get_execution_command().get_value(name, default_value)
+  if not value:
+    return UnknownConfiguration(name)
+  elif value == default_value:
+    return value
+
+  if expected_type == bool:
+    if isinstance(value, bool):
+      return value
+    elif isinstance(value, basestring):
+      if value != None and value.lower() == "true":
+        value = True
+      elif value != None and value.lower() == "false":
+        value = False
+      else:
+        raise Fail("Configuration {0} expected to be boolean (true or false), but found '{1}'".format(name, value))
+    else:
+      type_name = type(value).__name__
+      raise Fail(
+        "Configuration {0} expected to be boolean (true or false), but found instance of unknown type '{1}'".format(
+          name, type_name))
+  elif expected_type in [int, long, float]:
+    try:
+      value = expected_type(value)
+    except (ValueError, TypeError):
+      raise Fail("Configuration {0} expected to be number, but found '{1}'".format(name, value))
+  return value

ambari-common/src/test/python/TestExecutionCommand.py

@@ -32,6 +32,9 @@ class TestExecutionCommand(TestCase):
     try:
       with open(command_data_file) as f:
         self.__execution_command = execution_command.ExecutionCommand(json.load(f))
+        from resource_management.libraries.script import Script
+        Script.execution_command = self.__execution_command
+        Script.module_configs = Script.execution_command.get_module_configs()
     except IOError:
       Logger.error("Can not read json file with command parameters: ")
       sys.exit(1)
@@ -48,6 +51,10 @@ class TestExecutionCommand(TestCase):
     ganglia_server_hosts = self.__execution_command.get_component_hosts('ganglia_server')
     self.assertEqual(ganglia_server_hosts, 'host1')
 
+  def test_get_java_version(self):
+    java_version = self.__execution_command.get_java_version()
+    self.assertEqual(java_version, 8)
+
   def test_get_module_configs(self):
     module_configs = self.__execution_command.get_module_configs()
     self.assertNotEquals(module_configs, None)
@@ -66,13 +73,16 @@ class TestExecutionCommand(TestCase):
     self.assertEqual(int(properties.get('clientPort')), 2181)
     self.assertEqual(properties.get('fake'), None)
 
+    sqoop = bool(module_configs.get_all_properties("zookeeper", 'sqoop-env'))
+    self.assertFalse(sqoop)
+
   def test_get_stack_name(self):
     stack_name = self.__execution_command.get_mpack_name()
     self.assertEquals(stack_name, "HDPCORE")
 
   def test_access_to_module_configs(self):
     module_configs = self.__execution_command.get_module_configs()
-    is_zoo_cfg_there = module_configs.get_property_value("zookeeper", "zoo.cfg", "") is not None
+    is_zoo_cfg_there = bool(module_configs.get_all_properties("zookeeper", "zoo.cfg"))
     self.assertTrue(is_zoo_cfg_there)
-    zoo_cfg = module_configs.get_property_value("zookeeper", "zoo.cfg", "")
+    zoo_cfg = module_configs.get_all_properties("zookeeper", "zoo.cfg")
     self.assertTrue(isinstance(zoo_cfg, dict))

ambari-common/src/test/python/command.json

@@ -90,7 +90,8 @@
     "stack_version": "1.0.0-b141"
   },
   "ambariLevelParams": {
-    "host_sys_prepped": "false"
+    "host_sys_prepped": "false",
+    "java_version": "8"
   },
   "clusterHostInfo": {
     "ganglia_server_hosts": "host1",

ambari-server/src/main/resources/stack-hooks/after-INSTALL/scripts/params.py

@@ -89,9 +89,6 @@ namenode_opt_maxnewsize = module_configs.get_property_value(module_name, 'hadoop
 namenode_opt_permsize = format_jvm_option_value(module_configs.get_property_value(module_name, 'hadoop-env', 'namenode_opt_permsize', '128m'), '128m')
 namenode_opt_maxpermsize = format_jvm_option_value(module_configs.get_property_value(module_name, 'hadoop-env', 'namenode_opt_maxpermsize', '256m'), '256m')
 
-jtnode_opt_newsize = "200m"
-jtnode_opt_maxnewsize = "200m"
-jtnode_heapsize =  "1024m"
 ttnode_heapsize = "1024m"
 
 dtnode_heapsize = module_configs.get_property_value(module_name, 'hadoop-env', 'dtnode_heapsize')

ambari-server/src/main/resources/stack-hooks/before-ANY/scripts/params.py

@@ -65,6 +65,7 @@ sudo = AMBARI_SUDO_BINARY
 
 ambari_server_hostname = execution_command.get_ambari_server_host()
 
+stack_version_unformatted = execution_command.get_mpack_version()
 stack_version_formatted = execution_command.get_mpack_version()
 
 upgrade_type = Script.get_upgrade_type(execution_command.get_upgrade_type())
@@ -162,9 +163,6 @@ namenode_opt_maxnewsize = module_configs.get_property_value(module_name, 'hadoop
 namenode_opt_permsize = format_jvm_option_value(module_configs.get_property_value(module_name, 'hadoop-env', 'namenode_opt_permsize', '128m'), '128m')
 namenode_opt_maxpermsize = format_jvm_option_value(module_configs.get_property_value(module_name, 'hadoop-env', 'namenode_opt_maxpermsize', '256m'), '256m')
 
-jtnode_opt_newsize = "200m"
-jtnode_opt_maxnewsize = "200m"
-jtnode_heapsize =  "1024m"
 ttnode_heapsize = "1024m"
 
 dtnode_heapsize = module_configs.get_property_value(module_name, 'hadoop-env', 'dtnode_heapsize')

ambari-server/src/main/resources/stack-hooks/before-INSTALL/scripts/params.py

@@ -65,7 +65,6 @@ hcat_server_hosts = execution_command.get_component_hosts('webhcat_server')
 hive_server_host =  execution_command.get_component_hosts('hive_server')
 hbase_master_hosts = execution_command.get_component_hosts('hbase_master')
 hs_host = execution_command.get_component_hosts('historyserver')
-jtnode_host = execution_command.get_component_hosts('jtnode')
 namenode_host = execution_command.get_component_hosts('namenode')
 zk_hosts = execution_command.get_component_hosts('zookeeper_server')
 ganglia_server_hosts = execution_command.get_component_hosts('ganglia_server')
@@ -88,7 +87,6 @@ has_falcon_server = not len(falcon_host) == 0
 has_tez = bool(module_configs.get_all_properties(module_name, 'tez-site'))
 
 is_namenode_master = hostname in namenode_host
-is_jtnode_master = hostname in jtnode_host
 is_rmnode_master = hostname in rm_host
 is_hsnode_master = hostname in hs_host
 is_hbase_master = hostname in hbase_master_hosts

ambari-server/src/main/resources/stack-hooks/before-START/scripts/params.py

@@ -114,7 +114,6 @@ hcat_server_hosts = execution_command.get_component_hosts('webhcat_server')
 hive_server_host =  execution_command.get_component_hosts('hive_server')
 hbase_master_hosts = execution_command.get_component_hosts('hbase_master')
 hs_host = execution_command.get_component_hosts('historyserver')
-jtnode_host = execution_command.get_component_hosts('jtnode')
 namenode_host = execution_command.get_component_hosts('namenode')
 zk_hosts = execution_command.get_component_hosts('zookeeper_server')
 ganglia_server_hosts = execution_command.get_component_hosts('ganglia_server')
@@ -138,7 +137,6 @@ has_ganglia_server = not len(ganglia_server_hosts) == 0
 has_metric_collector = not len(ams_collector_hosts) == 0
 
 is_namenode_master = hostname in namenode_host
-is_jtnode_master = hostname in jtnode_host
 is_rmnode_master = hostname in rm_host
 is_hsnode_master = hostname in hs_host
 is_hbase_master = hostname in hbase_master_hosts
@@ -214,9 +212,6 @@ namenode_opt_maxnewsize = module_configs.get_property_value(module_name, 'hadoop
 namenode_opt_permsize = format_jvm_option_value(module_configs.get_property_value(module_name, 'hadoop-env', 'namenode_opt_permsize', '128m'), '128m')
 namenode_opt_maxpermsize = format_jvm_option_value(module_configs.get_property_value(module_name, 'hadoop-env', 'namenode_opt_maxpermsize', '256m'), '256m')
 
-jtnode_opt_newsize = "200m"
-jtnode_opt_maxnewsize = "200m"
-jtnode_heapsize =  "1024m"
 ttnode_heapsize = "1024m"
 
 dtnode_heapsize = module_configs.get_property_value(module_name, 'hadoop-env', 'dtnode_heapsize')
@@ -241,7 +236,7 @@ yarn_rm_summary_log_number_of_backup_files = module_configs.get_property_value(m
 
 #log4j.properties
 log4j_props = module_configs.get_property_value(module_name, 'hdfs-log4j', 'content')
-if log4j_props:
+if log4j_props and module_configs.get_property_value(module_name, 'yarn-log4j', 'content'):
   log4j_props += module_configs.get_property_value(module_name, 'yarn-log4j', 'content')
 
 refresh_topology = execution_command.need_refresh_topology()