AMBARI-12055. Some Yarn, Hive, Atlas, Oozie alerts fail on non-root in secured mode (aonishuk)

ambari-agent/src/main/python/ambari_agent/alerts/metric_alert.py

@@ -203,9 +203,10 @@ class MetricAlert(BaseAlert):
             tmp_dir = gettempdir()
 
           kerberos_executable_search_paths = self._get_configuration_value('{{kerberos-env/executable_search_paths}}')
+          smokeuser = self._get_configuration_value('{{cluster-env/smokeuser}}')
 
           response, error_msg, time_millis = curl_krb_request(tmp_dir, kerberos_keytab, kerberos_principal, url,
-                                          "metric_alert", kerberos_executable_search_paths, False, self.get_name())
+                                          "metric_alert", kerberos_executable_search_paths, False, self.get_name(), smokeuser)
           content = response
         else:
           url_opener = urllib2.build_opener(RefreshHeaderProcessor())
@@ -244,7 +245,7 @@ class MetricAlert(BaseAlert):
       if not json_is_valid and security_enabled and \
             kerberos_principal is not None and kerberos_keytab is not None:
         http_response_code, error_msg, time_millis = curl_krb_request(tmp_dir, kerberos_keytab, kerberos_principal, url,
-                                              "metric_alert", kerberos_executable_search_paths, True, self.get_name())
+                                              "metric_alert", kerberos_executable_search_paths, True, self.get_name(), smokeuser)
 
     return (value_list, http_response_code)
 

ambari-agent/src/main/python/ambari_agent/alerts/web_alert.py

@@ -164,7 +164,10 @@ class WebAlert(BaseAlert):
       if self.uri_property_keys.kerberos_keytab is not None:
         kerberos_keytab = self._get_configuration_value(self.uri_property_keys.kerberos_keytab)
 
-      if kerberos_principal is not None and kerberos_keytab is not None:
+      security_enabled = self._get_configuration_value('{{cluster-env/security_enabled}}')
+      
+      if kerberos_principal is not None and kerberos_keytab is not None \
+        and security_enabled is not None and security_enabled.lower() == "true":
         # Create the kerberos credentials cache (ccache) file and set it in the environment to use
         # when executing curl. Use the md5 hash of the combination of the principal and keytab file
         # to generate a (relatively) unique cache filename so that we can use it as needed.
@@ -174,9 +177,10 @@ class WebAlert(BaseAlert):
 
         # Get the configured Kerberos executables search paths, if any
         kerberos_executable_search_paths = self._get_configuration_value('{{kerberos-env/executable_search_paths}}')
+        smokeuser = self._get_configuration_value('{{cluster-env/smokeuser}}')
 
         response_code, error_msg, time_millis = curl_krb_request(tmp_dir, kerberos_keytab, kerberos_principal, url,
-                                               "web_alert", kerberos_executable_search_paths, True, self.get_name())
+                                               "web_alert", kerberos_executable_search_paths, True, self.get_name(), smokeuser)
       else:
         # kerberos is not involved; use urllib2
         response_code, time_millis, error_msg = self._make_web_request_urllib(url)

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

@@ -26,6 +26,8 @@ import os
 import time
 import subprocess
 
+from resource_management.core import shell
+from resource_management.core.exceptions import Fail
 from get_kinit_path import get_kinit_path
 from get_klist_path import get_klist_path
 # hashlib is supplied as of Python 2.5 as the replacement interface for md5
@@ -45,7 +47,7 @@ logger = logging.getLogger()
 
 
 def curl_krb_request(tmp_dir, keytab, principal, url, cache_file_prefix, krb_exec_search_paths,
-                     return_only_http_code, alert_name):
+                     return_only_http_code, alert_name, user):
   import uuid
   # Create the kerberos credentials cache (ccache) file and set it in the environment to use
   # when executing curl. Use the md5 hash of the combination of the principal and keytab file
@@ -61,7 +63,7 @@ def curl_krb_request(tmp_dir, keytab, principal, url, cache_file_prefix, krb_exe
   else:
     klist_path_local = get_klist_path()
 
-  if os.system("{0} -s {1}".format(klist_path_local, ccache_file_path)) != 0:
+  if shell.call("{0} -s {1}".format(klist_path_local, ccache_file_path), user=user)[0] != 0:
     if krb_exec_search_paths:
       kinit_path_local = get_kinit_path(krb_exec_search_paths)
     else:
@@ -69,7 +71,7 @@ def curl_krb_request(tmp_dir, keytab, principal, url, cache_file_prefix, krb_exe
     logger.debug("[Alert][{0}] Enabling Kerberos authentication via GSSAPI using ccache at {1}.".format(
       alert_name, ccache_file_path))
 
-    os.system("{0} -l 5m -c {1} -kt {2} {3} > /dev/null".format(kinit_path_local, ccache_file_path, keytab, principal))
+    shell.checked_call("{0} -l 5m -c {1} -kt {2} {3} > /dev/null".format(kinit_path_local, ccache_file_path, keytab, principal), user=user)
   else:
     logger.debug("[Alert][{0}] Kerberos authentication via GSSAPI already enabled using ccache at {1}.".format(
       alert_name, ccache_file_path))
@@ -87,20 +89,18 @@ def curl_krb_request(tmp_dir, keytab, principal, url, cache_file_prefix, krb_exe
   error_msg = None
   try:
     if return_only_http_code:
-      curl = subprocess.Popen(['curl', '-k', '--negotiate', '-u', ':', '-b', cookie_file, '-c', cookie_file, '-w',
+      _, curl_stdout, curl_stderr = shell.checked_call(['curl', '-k', '--negotiate', '-u', ':', '-b', cookie_file, '-c', cookie_file, '-w',
                              '%{http_code}', url, '--connect-timeout', str(CONNECTION_TIMEOUT),'-o', '/dev/null'],
-                             stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=kerberos_env)
+                             stderr=subprocess.PIPE, env=kerberos_env, user=user)
     else:
       # returns response body
-      curl = subprocess.Popen(['curl', '-k', '--negotiate', '-u', ':', '-b', cookie_file, '-c', cookie_file,
+      _, curl_stdout, curl_stderr = shell.checked_call(['curl', '-k', '--negotiate', '-u', ':', '-b', cookie_file, '-c', cookie_file,
                              url, '--connect-timeout', str(CONNECTION_TIMEOUT)],
-                             stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=kerberos_env)
-
-    curl_stdout, curl_stderr = curl.communicate()
-  except Exception, exception:
+                             stderr=subprocess.PIPE, env=kerberos_env, user=user)
+  except Fail:
     if logger.isEnabledFor(logging.DEBUG):
       logger.exception("[Alert][{0}] Unable to make a web request.".format(alert_name))
-    raise Exception(exception)
+    raise
   finally:
     if os.path.isfile(cookie_file):
       os.remove(cookie_file)

ambari-server/src/main/resources/common-services/HDFS/2.1.0.2.0/package/alerts/alert_checkpoint_time.py

@@ -49,6 +49,7 @@ CONNECTION_TIMEOUT_DEFAULT = 5.0
 KERBEROS_KEYTAB = '{{hdfs-site/dfs.web.authentication.kerberos.keytab}}'
 KERBEROS_PRINCIPAL = '{{hdfs-site/dfs.web.authentication.kerberos.principal}}'
 SECURITY_ENABLED_KEY = '{{cluster-env/security_enabled}}'
+SMOKEUSER_KEY = "{{cluster-env/smokeuser}}"
 
 logger = logging.getLogger()
 
@@ -58,7 +59,7 @@ def get_tokens():
   to build the dictionary passed into execute
   """
   return (NN_HTTP_ADDRESS_KEY, NN_HTTPS_ADDRESS_KEY, NN_HTTP_POLICY_KEY, 
-      NN_CHECKPOINT_TX_KEY, NN_CHECKPOINT_PERIOD_KEY, KERBEROS_KEYTAB, KERBEROS_PRINCIPAL, SECURITY_ENABLED_KEY)
+      NN_CHECKPOINT_TX_KEY, NN_CHECKPOINT_PERIOD_KEY, KERBEROS_KEYTAB, KERBEROS_PRINCIPAL, SECURITY_ENABLED_KEY, SMOKEUSER_KEY)
   
 
 def execute(configurations={}, parameters={}, host_name=None):
@@ -96,6 +97,9 @@ def execute(configurations={}, parameters={}, host_name=None):
 
   if NN_CHECKPOINT_PERIOD_KEY in configurations:
     checkpoint_period = configurations[NN_CHECKPOINT_PERIOD_KEY]
+    
+  if SMOKEUSER_KEY in configurations:
+    smokeuser = configurations[SMOKEUSER_KEY]
 
   security_enabled = False
   if SECURITY_ENABLED_KEY in configurations:
@@ -145,13 +149,13 @@ def execute(configurations={}, parameters={}, host_name=None):
       env = Environment.get_instance()
       last_checkpoint_time_response, error_msg, time_millis = curl_krb_request(env.tmp_dir, kerberos_keytab,
                                     kerberos_principal, last_checkpoint_time_qry,"checkpoint_time_alert", None, False,
-                                    "NameNode Last Checkpoint")
+                                    "NameNode Last Checkpoint", smokeuser)
       last_checkpoint_time_response_json = json.loads(last_checkpoint_time_response)
       last_checkpoint_time = int(last_checkpoint_time_response_json["beans"][0]["LastCheckpointTime"])
 
       journal_transaction_info_response, error_msg, time_millis = curl_krb_request(env.tmp_dir, kerberos_keytab,
                                       kerberos_principal, journal_transaction_info_qry,"checkpoint_time_alert", None,
-                                      False, "NameNode Last Checkpoint")
+                                      False, "NameNode Last Checkpoint", smokeuser)
       journal_transaction_info_response_json = json.loads(journal_transaction_info_response)
       journal_transaction_info = journal_transaction_info_response_json["beans"][0]["JournalTransactionInfo"]
     else:

ambari-server/src/main/resources/common-services/HDFS/2.1.0.2.0/package/alerts/alert_ha_namenode_health.py

@@ -42,6 +42,7 @@ DFS_POLICY_KEY = '{{hdfs-site/dfs.http.policy}}'
 KERBEROS_KEYTAB = '{{hdfs-site/dfs.web.authentication.kerberos.keytab}}'
 KERBEROS_PRINCIPAL = '{{hdfs-site/dfs.web.authentication.kerberos.principal}}'
 SECURITY_ENABLED_KEY = '{{cluster-env/security_enabled}}'
+SMOKEUSER_KEY = '{{cluster-env/smokeuser}}'
 
 CONNECTION_TIMEOUT_KEY = 'connection.timeout'
 CONNECTION_TIMEOUT_DEFAULT = 5.0
@@ -54,7 +55,7 @@ def get_tokens():
   to build the dictionary passed into execute
   """
   return (HDFS_SITE_KEY, NAMESERVICE_KEY, NN_HTTP_ADDRESS_KEY,
-  NN_HTTPS_ADDRESS_KEY, DFS_POLICY_KEY, KERBEROS_KEYTAB, KERBEROS_PRINCIPAL, SECURITY_ENABLED_KEY)
+  NN_HTTPS_ADDRESS_KEY, DFS_POLICY_KEY, SMOKEUSER_KEY, KERBEROS_KEYTAB, KERBEROS_PRINCIPAL, SECURITY_ENABLED_KEY)
   
 
 def execute(configurations={}, parameters={}, host_name=None):
@@ -76,6 +77,9 @@ def execute(configurations={}, parameters={}, host_name=None):
   # hdfs-site is required
   if not HDFS_SITE_KEY in configurations:
     return (RESULT_STATE_UNKNOWN, ['{0} is a required parameter for the script'.format(HDFS_SITE_KEY)])
+  
+  if SMOKEUSER_KEY in configurations:
+    smokeuser = configurations[SMOKEUSER_KEY]
 
   # parse script arguments
   connection_timeout = CONNECTION_TIMEOUT_DEFAULT
@@ -140,7 +144,7 @@ def execute(configurations={}, parameters={}, host_name=None):
           env = Environment.get_instance()
           state_response, error_msg, time_millis  = curl_krb_request(env.tmp_dir, kerberos_keytab, kerberos_principal,
                                                     jmx_uri,"ha_nn_health", None, False,
-                                                    "NameNode High Availability Health")
+                                                    "NameNode High Availability Health", smokeuser)
           state_response_json = json.loads(state_response)
           state = state_response_json["beans"][0]['State']
         else:

ambari-server/src/main/resources/common-services/HDFS/2.1.0.2.0/package/scripts/namenode.py

@@ -26,6 +26,7 @@ import ambari_simplejson as json # simplejson is much faster comparing to Python
 
 from resource_management import Script
 from resource_management.core.resources.system import Execute
+from resource_management.core import shell
 from resource_management.libraries.functions import conf_select
 from resource_management.libraries.functions import hdp_select
 from resource_management.libraries.functions.version import compare_versions, format_hdp_stack_version
@@ -211,7 +212,7 @@ class NameNodeDefault(NameNode):
       # is in the cache
       klist_cmd = format("{klist_path_local} -s {ccache_file_path}")
       kinit_cmd = format("{kinit_path_local} -c {ccache_file_path} -kt {hdfs_user_keytab} {hdfs_principal_name}")
-      if os.system(klist_cmd) != 0:
+      if shell.call(klist_cmd, user=params.hdfs_user)[0] != 0:
         Execute(kinit_cmd, user=params.hdfs_user)
 
     def calculateCompletePercent(first, current):

ambari-server/src/main/resources/common-services/HIVE/0.12.0.2.0/package/alerts/alert_webhcat_server.py

@@ -26,7 +26,7 @@ import urllib2
 
 from resource_management.core.environment import Environment
 from resource_management.core.resources import Execute
-from resource_management.core.shell import call
+from resource_management.core import shell
 from resource_management.libraries.functions import format
 from resource_management.libraries.functions import get_kinit_path
 from resource_management.libraries.functions import get_klist_path
@@ -46,6 +46,7 @@ TEMPLETON_PORT_KEY = '{{webhcat-site/templeton.port}}'
 SECURITY_ENABLED_KEY = '{{cluster-env/security_enabled}}'
 WEBHCAT_PRINCIPAL_KEY = '{{webhcat-site/templeton.kerberos.principal}}'
 WEBHCAT_KEYTAB_KEY = '{{webhcat-site/templeton.kerberos.keytab}}'
+SMOKEUSER_KEY = '{{cluster-env/smokeuser}}'
 
 # The configured Kerberos executable search paths, if any
 KERBEROS_EXECUTABLE_SEARCH_PATHS_KEY = '{{kerberos-env/executable_search_paths}}'
@@ -58,7 +59,6 @@ CONNECTION_TIMEOUT_DEFAULT = 5.0
 CURL_CONNECTION_TIMEOUT_DEFAULT = str(int(CONNECTION_TIMEOUT_DEFAULT))
 
 # default smoke user
-SMOKEUSER_KEY = "{{cluster-env/smokeuser}}"
 SMOKEUSER_SCRIPT_PARAM_KEY = 'default.smoke.user'
 SMOKEUSER_DEFAULT = 'ambari-qa'
 
@@ -111,9 +111,6 @@ def execute(configurations={}, parameters={}, host_name=None):
   if SMOKEUSER_KEY in configurations:
     smokeuser = configurations[SMOKEUSER_KEY]
 
-  if SMOKEUSER_SCRIPT_PARAM_KEY in parameters:
-    smokeuser = parameters[SMOKEUSER_SCRIPT_PARAM_KEY]
-
   # webhcat always uses http, never SSL
   query_url = "http://{0}:{1}/templeton/v1/status?user.name={2}".format(host_name, webhcat_port, smokeuser)
 
@@ -150,20 +147,20 @@ def execute(configurations={}, parameters={}, host_name=None):
       # Determine if we need to kinit by testing to see if the relevant cache exists and has
       # non-expired tickets.  Tickets are marked to expire after 5 minutes to help reduce the number
       # it kinits we do but recover quickly when keytabs are regenerated
-      return_code, _ = call(klist_command)
+      return_code, _ = shell.call(klist_command, user=smokeuser)
       if return_code != 0:
         kinit_path_local = get_kinit_path(kerberos_executable_search_paths)
         kinit_command = format("{kinit_path_local} -l 5m -c {ccache_file} -kt {webhcat_keytab} {webhcat_principal}; ")
 
         # kinit so that curl will work with --negotiate
-        Execute(kinit_command)
+        Execute(kinit_command,
+                user=smokeuser,
+        )
 
       # make a single curl call to get just the http code
-      curl = subprocess.Popen(['curl', '--negotiate', '-u', ':', '-sL', '-w',
+      _, stdout, stderr = shell.checked_call(['curl', '--negotiate', '-u', ':', '-sL', '-w',
         '%{http_code}', '--connect-timeout', curl_connection_timeout,
-        '-o', '/dev/null', query_url], stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=kerberos_env)
-
-      stdout, stderr = curl.communicate()
+        '-o', '/dev/null', query_url], stderr=subprocess.PIPE, env=kerberos_env)
 
       if stderr != '':
         raise Exception(stderr)
@@ -183,11 +180,10 @@ def execute(configurations={}, parameters={}, host_name=None):
 
       # now that we have the http status and it was 200, get the content
       start_time = time.time()
-      curl = subprocess.Popen(['curl', '--negotiate', '-u', ':', '-sL',
+      _, stdout, stderr = shell.checked_call(['curl', '--negotiate', '-u', ':', '-sL',
         '--connect-timeout', curl_connection_timeout, query_url, ],
-        stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=kerberos_env)
+        stderr=subprocess.PIPE, env=kerberos_env)
 
-      stdout, stderr = curl.communicate()
       total_time = time.time() - start_time
 
       if stderr != '':

ambari-server/src/main/resources/common-services/YARN/2.1.0.2.0/package/alerts/alert_nodemanager_health.py

@@ -43,6 +43,7 @@ CRITICAL_NODEMANAGER_UNKNOWN_JSON_MESSAGE = 'Unable to determine NodeManager hea
 KERBEROS_KEYTAB = '{{yarn-site/yarn.nodemanager.webapp.spnego-keytab-file}}'
 KERBEROS_PRINCIPAL = '{{yarn-site/yarn.nodemanager.webapp.spnego-principal}}'
 SECURITY_ENABLED_KEY = '{{cluster-env/security_enabled}}'
+SMOKEUSER_KEY = '{{cluster-env/smokeuser}}'
 
 NODEMANAGER_DEFAULT_PORT = 8042
 
@@ -55,7 +56,7 @@ def get_tokens():
   to build the dictionary passed into execute
   """
   return (NODEMANAGER_HTTP_ADDRESS_KEY,NODEMANAGER_HTTPS_ADDRESS_KEY,
-  YARN_HTTP_POLICY_KEY, KERBEROS_KEYTAB, KERBEROS_PRINCIPAL, SECURITY_ENABLED_KEY)
+  YARN_HTTP_POLICY_KEY, SMOKEUSER_KEY, KERBEROS_KEYTAB, KERBEROS_PRINCIPAL, SECURITY_ENABLED_KEY)
   
 
 def execute(configurations={}, parameters={}, host_name=None):
@@ -77,6 +78,9 @@ def execute(configurations={}, parameters={}, host_name=None):
   https_uri = None
   http_policy = 'HTTP_ONLY'
 
+  if SMOKEUSER_KEY in configurations:
+    smokeuser = configurations[SMOKEUSER_KEY]
+    
   security_enabled = False
   if SECURITY_ENABLED_KEY in configurations:
     security_enabled = str(configurations[SECURITY_ENABLED_KEY]).upper() == 'TRUE'
@@ -138,7 +142,7 @@ def execute(configurations={}, parameters={}, host_name=None):
     if kerberos_principal is not None and kerberos_keytab is not None and security_enabled:
       env = Environment.get_instance()
       url_response, error_msg, time_millis  = curl_krb_request(env.tmp_dir, kerberos_keytab, kerberos_principal,
-                                                query, "nm_health_alert", None, False, "NodeManager Health")
+                                                query, "nm_health_alert", None, False, "NodeManager Health", smokeuser)
 
       json_response = json.loads(url_response)
     else:

ambari-server/src/main/resources/common-services/YARN/2.1.0.2.0/package/alerts/alert_nodemanagers_summary.py

@@ -36,6 +36,7 @@ YARN_HTTP_POLICY_KEY = '{{yarn-site/yarn.http.policy}}'
 KERBEROS_KEYTAB = '{{yarn-site/yarn.nodemanager.webapp.spnego-keytab-file}}'
 KERBEROS_PRINCIPAL = '{{yarn-site/yarn.nodemanager.webapp.spnego-principal}}'
 SECURITY_ENABLED_KEY = '{{cluster-env/security_enabled}}'
+SMOKEUSER_KEY = '{{cluster-env/smokeuser}}'
 
 CONNECTION_TIMEOUT_KEY = 'connection.timeout'
 CONNECTION_TIMEOUT_DEFAULT = 5.0
@@ -48,7 +49,7 @@ def get_tokens():
   to build the dictionary passed into execute
   """
   return NODEMANAGER_HTTP_ADDRESS_KEY, NODEMANAGER_HTTPS_ADDRESS_KEY, \
-    YARN_HTTP_POLICY_KEY, KERBEROS_KEYTAB, KERBEROS_PRINCIPAL, SECURITY_ENABLED_KEY
+    YARN_HTTP_POLICY_KEY, SMOKEUSER_KEY, KERBEROS_KEYTAB, KERBEROS_PRINCIPAL, SECURITY_ENABLED_KEY
 
 
 def execute(configurations={}, parameters={}, host_name=None):
@@ -90,6 +91,9 @@ def execute(configurations={}, parameters={}, host_name=None):
 
   if YARN_HTTP_POLICY_KEY in configurations:
     http_policy = configurations[YARN_HTTP_POLICY_KEY]
+    
+  if SMOKEUSER_KEY in configurations:
+    smokeuser = configurations[SMOKEUSER_KEY]
 
   # parse script arguments
   connection_timeout = CONNECTION_TIMEOUT_DEFAULT
@@ -113,7 +117,7 @@ def execute(configurations={}, parameters={}, host_name=None):
       env = Environment.get_instance()
       url_response, error_msg, time_millis  = curl_krb_request(env.tmp_dir, kerberos_keytab, kerberos_principal,
                                               live_nodemanagers_qry, "nm_health_summary_alert", None, False,
-                                              "NodeManager Health Summary")
+                                              "NodeManager Health Summary", smokeuser)
       try:
         url_response_json = json.loads(url_response)
         live_nodemanagers = json.loads(url_response_json["beans"][0]["LiveNodeManagers"])
@@ -126,7 +130,7 @@ def execute(configurations={}, parameters={}, host_name=None):
       if convert_to_json_failed:
         response_code, error_msg, time_millis  = curl_krb_request(env.tmp_dir, kerberos_keytab, kerberos_principal,
                                                     live_nodemanagers_qry, "nm_health_summary_alert", None, True,
-                                                    "NodeManager Health Summary")
+                                                    "NodeManager Health Summary", smokeuser)
     else:
       live_nodemanagers = json.loads(get_value_from_jmx(live_nodemanagers_qry,
       "LiveNodeManagers", connection_timeout))

ambari-server/src/test/python/stacks/2.0.6/HDFS/test_namenode.py

@@ -983,12 +983,14 @@ class TestNamenode(RMFTestCase):
                        command = "rebalancehdfs",
                        config_file = "rebalancehdfs_secured.json",
                        hdp_stack_version = self.STACK_VERSION,
-                       target = RMFTestCase.TARGET_COMMON_SERVICES
+                       target = RMFTestCase.TARGET_COMMON_SERVICES,
+                       call_mocks=[(1, "no kinit")]
     )
     tempdir = tempfile.gettempdir()
     ccache_path =  os.path.join(tempfile.gettempdir(), "hdfs_rebalance_cc_7add60ca651f1bd1ed909a6668937ba9")
     kinit_cmd = "/usr/bin/kinit -c {0} -kt /etc/security/keytabs/hdfs.headless.keytab hdfs@EXAMPLE.COM".format(ccache_path)
     rebalance_cmd = "ambari-sudo.sh su hdfs -l -s /bin/bash -c 'export  PATH=/bin:/usr/bin KRB5CCNAME={0} ; hdfs --config /etc/hadoop/conf balancer -threshold -1'".format(ccache_path)
+
     self.assertResourceCalled('Execute', kinit_cmd,
                               user = 'hdfs',
                               )