AMBARI-9882. Pig Service check fails in a Kerberized cluster (alejandro)

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

@@ -73,10 +73,11 @@ def _get_tar_source_and_dest_folder(tarball_prefix):
   return component_tar_source_file, component_tar_destination_folder
 
 
-def _copy_files(source_and_dest_pairs, file_owner, group_owner, kinit_if_needed):
+def _copy_files(source_and_dest_pairs, component_user, file_owner, group_owner, kinit_if_needed):
   """
   :param source_and_dest_pairs: List of tuples (x, y), where x is the source file in the local file system,
   and y is the destination file path in HDFS
+  :param component_user:  User that will execute the Hadoop commands, usually smokeuser
   :param file_owner: Owner to set for the file copied to HDFS (typically hdfs account)
   :param group_owner: Owning group to set for the file copied to HDFS (typically hadoop group)
   :param kinit_if_needed: kinit command if it is needed, otherwise an empty string
@@ -96,6 +97,7 @@ def _copy_files(source_and_dest_pairs, file_owner, group_owner, kinit_if_needed)
         params.HdfsDirectory(destination_dir,
                              action="create",
                              owner=file_owner,
+                             hdfs_user=params.hdfs_user,   # this will be the user to run the commands as
                              mode=0555
         )
 
@@ -103,13 +105,15 @@ def _copy_files(source_and_dest_pairs, file_owner, group_owner, kinit_if_needed)
                       mode=0444,
                       owner=file_owner,
                       group=group_owner,
+                      user=params.hdfs_user,               # this will be the user to run the commands as
                       dest_dir=destination_dir,
                       kinnit_if_needed=kinit_if_needed,
                       hdfs_user=params.hdfs_user,
                       hadoop_bin_dir=params.hadoop_bin_dir,
                       hadoop_conf_dir=params.hadoop_conf_dir
         )
-      except:
+      except Exception, e:
+        Logger.error("Failed to copy file. Source: %s, Destination: %s. Error: %s" % (source, destination, e.message))
         return_value = 1
   return return_value
 
@@ -118,8 +122,8 @@ def copy_tarballs_to_hdfs(tarball_prefix, hdp_select_component_name, component_u
   """
   :param tarball_prefix: Prefix of the tarball must be one of tez, hive, mr, pig
   :param hdp_select_component_name: Component name to get the status to determine the version
-  :param component_user: User that will execute the Hadoop commands
-  :param file_owner: Owner of the files copied to HDFS (typically hdfs account)
+  :param component_user: User that will execute the Hadoop commands, usually smokeuser
+  :param file_owner: Owner of the files copied to HDFS (typically hdfs user)
   :param group_owner: Group owner of the files copied to HDFS (typically hadoop group)
   :return: Returns 0 on success, 1 if no files were copied, and in some cases may raise an exception.
 
@@ -192,5 +196,5 @@ def copy_tarballs_to_hdfs(tarball_prefix, hdp_select_component_name, component_u
 
   if not does_hdfs_file_exist:
     source_and_dest_pairs = [(component_tar_source_file, destination_file), ]
-    return _copy_files(source_and_dest_pairs, file_owner, group_owner, kinit_if_needed)
+    return _copy_files(source_and_dest_pairs, component_user, file_owner, group_owner, kinit_if_needed)
   return 1

ambari-common/src/main/python/resource_management/libraries/providers/copy_from_local.py

@@ -34,6 +34,7 @@ class CopyFromLocalProvider(Provider):
     dest_dir = self.resource.dest_dir
     dest_file = self.resource.dest_file
     kinnit_if_needed = self.resource.kinnit_if_needed
+    user = self.resource.user   # user to perform commands as. If not provided, default to the owner
     owner = self.resource.owner
     group = self.resource.group
     mode = self.resource.mode
@@ -53,14 +54,14 @@ class CopyFromLocalProvider(Provider):
     
     if kinnit_if_needed:
       Execute(kinnit_if_needed, 
-              user=owner,
+              user=user if user else owner,
       )
     
-    unless_cmd = as_user(format("PATH=$PATH:{bin_dir} hadoop fs -ls {dest_path}"), owner)
+    unless_cmd = as_user(format("PATH=$PATH:{bin_dir} hadoop fs -ls {dest_path}"), user if user else owner)
 
     ExecuteHadoop(copy_cmd,
                   not_if=unless_cmd,
-                  user=owner,
+                  user=user if user else owner,
                   bin_dir=bin_dir,
                   conf_dir=hadoop_conf_path
                   )

ambari-common/src/main/python/resource_management/libraries/resources/copy_from_local.py

@@ -29,10 +29,11 @@ class CopyFromLocal(Resource):
   path = ResourceArgument(default=lambda obj: obj.name)
   dest_dir = ResourceArgument(required=True)
   dest_file = ResourceArgument()
-  owner = ResourceArgument(required=True)
-  group = ResourceArgument()
-  mode = ResourceArgument()
+  owner = ResourceArgument(required=True)  # file user owner
+  group = ResourceArgument()               # file group user
+  mode = ResourceArgument()                # file ACL mode
   kinnit_if_needed = ResourceArgument(default='')
+  user = ResourceArgument()                # user to perform commands as. If not provided, default to the owner
   hadoop_conf_dir = ResourceArgument(default='/etc/hadoop/conf')
   hdfs_user = ResourceArgument(default='hdfs')
   hadoop_bin_dir = ResourceArgument(default='/usr/bin')

ambari-server/src/main/resources/common-services/PIG/0.12.0.2.0/package/scripts/params.py

@@ -70,7 +70,7 @@ import functools
 HdfsDirectory = functools.partial(
   HdfsDirectory,
   conf_dir=hadoop_conf_dir,
-  hdfs_user=hdfs_principal_name if security_enabled else hdfs_user,
+  hdfs_user=hdfs_user,
   security_enabled = security_enabled,
   keytab = hdfs_user_keytab,
   kinit_path_local = kinit_path_local,

ambari-server/src/main/resources/common-services/PIG/0.12.0.2.0/package/scripts/service_check.py

@@ -91,14 +91,14 @@ class PigServiceCheck(Script):
                 user=params.smokeuser
         )
 
-      Execute( format("pig -x tez {tmp_dir}/pigSmoke.sh"),
+      Execute(format("pig -x tez {tmp_dir}/pigSmoke.sh"),
         tries     = 3,
         try_sleep = 5,
         path      = format('{pig_bin_dir}:/usr/sbin:/sbin:/usr/local/bin:/bin:/usr/bin'),
         user      = params.smokeuser
       )
 
-      ExecuteHadoop( test_cmd,
+      ExecuteHadoop(test_cmd,
         user      = params.smokeuser,
         conf_dir = params.hadoop_conf_dir,
         bin_dir = params.hadoop_bin_dir

ambari-server/src/main/resources/common-services/SPARK/1.2.0.2.2/package/scripts/params.py

@@ -145,7 +145,7 @@ import functools
 HdfsDirectory = functools.partial(
   HdfsDirectory,
   conf_dir=hadoop_conf_dir,
-  hdfs_user=hdfs_principal_name if security_enabled else hdfs_user,
+  hdfs_user=hdfs_user,
   security_enabled = security_enabled,
   keytab = hdfs_user_keytab,
   kinit_path_local = kinit_path_local,

ambari-server/src/main/resources/common-services/TEZ/0.4.0.2.1/package/scripts/params.py

@@ -61,7 +61,7 @@ import functools
 HdfsDirectory = functools.partial(
   HdfsDirectory,
   conf_dir=hadoop_conf_dir,
-  hdfs_user=hdfs_principal_name if security_enabled else hdfs_user,
+  hdfs_user=hdfs_user,
   security_enabled=security_enabled,
   keytab=hdfs_user_keytab,
   kinit_path_local=kinit_path_local,

ambari-server/src/test/python/stacks/2.2/PIG/test_pig_service_check.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python
+
+'''
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+'''
+from mock.mock import patch, MagicMock
+
+from stacks.utils.RMFTestCase import *
+from resource_management.libraries.functions import dynamic_variable_interpretation
+
+
+class TestPigServiceCheck(RMFTestCase):
+  COMMON_SERVICES_PACKAGE_DIR = "PIG/0.12.0.2.0/package"
+  STACK_VERSION = "2.2"
+
+  @patch.object(dynamic_variable_interpretation, "copy_tarballs_to_hdfs")
+  def test_service_check_secure(self, copy_tarball_mock):
+    copy_tarball_mock.return_value = 0
+
+    self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/service_check.py",
+                       classname="PigServiceCheck",
+                       command="service_check",
+                       config_file="pig-service-check-secure.json",
+                       hdp_stack_version=self.STACK_VERSION,
+                       target=RMFTestCase.TARGET_COMMON_SERVICES
+    )
+    self.assertResourceCalled("ExecuteHadoop", "dfs -rmr pigsmoke.out passwd; hadoop --config /etc/hadoop/conf dfs -put /etc/passwd passwd ",
+      try_sleep=5,
+      tries=3,
+      user="ambari-qa",
+      conf_dir="/etc/hadoop/conf",
+      security_enabled=True,
+      principal="ambari-qa@EXAMPLE.COM",
+      keytab="/etc/security/keytabs/smokeuser.headless.keytab",
+      bin_dir="/usr/hdp/current/hadoop-client/bin",
+      kinit_path_local="/usr/bin/kinit"
+    )
+
+    self.assertResourceCalled("File", "/tmp/pigSmoke.sh",
+      content=StaticFile("pigSmoke.sh"),
+      mode=0755
+    )
+
+    self.assertResourceCalled("Execute", "pig /tmp/pigSmoke.sh",
+      path=["/usr/hdp/current/pig-client/bin:/usr/sbin:/sbin:/usr/local/bin:/bin:/usr/bin"],
+      tries=3,
+      user="ambari-qa",
+      try_sleep=5
+    )
+
+    self.assertResourceCalled("ExecuteHadoop", "fs -test -e pigsmoke.out",
+      user="ambari-qa",
+      bin_dir="/usr/hdp/current/hadoop-client/bin",
+      conf_dir="/etc/hadoop/conf"
+    )
+
+    # Specific to HDP 2.2 and kerberized cluster
+    self.assertResourceCalled("ExecuteHadoop", "dfs -rmr pigsmoke.out passwd; hadoop --config /etc/hadoop/conf dfs -put /etc/passwd passwd ",
+      tries=3,
+      try_sleep=5,
+      user="ambari-qa",
+      conf_dir="/etc/hadoop/conf",
+      keytab="/etc/security/keytabs/smokeuser.headless.keytab",
+      principal="ambari-qa@EXAMPLE.COM",
+      security_enabled=True,
+      kinit_path_local="/usr/bin/kinit",
+      bin_dir="/usr/hdp/current/hadoop-client/bin"
+    )
+
+    copy_tarball_mock.assert_called_once_with("tez", "hadoop-client", "ambari-qa", "hdfs", "hadoop")
+
+    self.assertResourceCalled("Execute", "/usr/bin/kinit -kt /etc/security/keytabs/smokeuser.headless.keytab ambari-qa@EXAMPLE.COM;",
+      user="ambari-qa")
+
+    self.assertResourceCalled("Execute", "pig -x tez /tmp/pigSmoke.sh",
+      tries=3,
+      try_sleep=5,
+      path=["/usr/hdp/current/pig-client/bin:/usr/sbin:/sbin:/usr/local/bin:/bin:/usr/bin"],
+      user="ambari-qa"
+    )
+
+    self.assertResourceCalled("ExecuteHadoop", "fs -test -e pigsmoke.out",
+      user="ambari-qa",
+      bin_dir="/usr/hdp/current/hadoop-client/bin",
+      conf_dir="/etc/hadoop/conf"
+    )
+    self.assertNoMoreResources()
+