AMBARI-16042. Add ability to log .out files into the agent log (aonishuk)

ambari-agent/src/main/python/ambari_agent/CustomServiceOrchestrator.py

@@ -189,9 +189,11 @@ class CustomServiceOrchestrator():
 
       python_executor = self.get_py_executor(forced_command_name)
       backup_log_files = not command_name in self.DONT_BACKUP_LOGS_FOR_COMMANDS
+      log_out_files = self.config.get("logging","log_out_files", default="0") != "0"
+      
       for py_file, current_base_dir in filtered_py_file_list:
         log_info_on_failure = not command_name in self.DONT_DEBUG_FAILURES_FOR_COMMANDS
-        script_params = [command_name, json_path, current_base_dir, tmpstrucoutfile, logger_level, self.exec_tmp_dir]
+        script_params = [command_name, json_path, current_base_dir, tmpstrucoutfile, logger_level, self.exec_tmp_dir, str(log_out_files)]
         ret = python_executor.run_file(py_file, script_params,
                                tmpoutfile, tmperrfile, timeout,
                                tmpstrucoutfile, self.map_task_to_process,

ambari-agent/src/test/python/ambari_agent/TestCustomServiceOrchestrator.py

@@ -46,6 +46,7 @@ from only_for_platform import get_platform, os_distro_value, PLATFORM_WINDOWS
 
 class TestCustomServiceOrchestrator(TestCase):
 
+
   def setUp(self):
     # disable stdout
     out = StringIO.StringIO()
@@ -54,6 +55,7 @@ class TestCustomServiceOrchestrator(TestCase):
     tmpdir = tempfile.gettempdir()
     exec_tmp_dir = os.path.join(tmpdir, 'tmp')
     self.config = ConfigParser.RawConfigParser()
+    self.config.get = AmbariConfig().get
     self.config.add_section('agent')
     self.config.set('agent', 'prefix', tmpdir)
     self.config.set('agent', 'cache_dir', "/cachedir")
@@ -453,13 +455,15 @@ class TestCustomServiceOrchestrator(TestCase):
     self.assertEqual(runningCommand['status'], ActionQueue.FAILED_STATUS)
 
 
+  @patch.object(AmbariConfig, "get")
   @patch.object(CustomServiceOrchestrator, "dump_command_to_json")
   @patch.object(PythonExecutor, "run_file")
   @patch.object(FileCache, "__init__")
   @patch.object(FileCache, "get_custom_actions_base_dir")
   def test_runCommand_custom_action(self, get_custom_actions_base_dir_mock,
                                     FileCache_mock,
-                                    run_file_mock, dump_command_to_json_mock):
+                                    run_file_mock, dump_command_to_json_mock, ambari_config_get):
+    ambari_config_get.return_value = "0"
     FileCache_mock.return_value = None
     get_custom_actions_base_dir_mock.return_value = "some path"
     _, script = tempfile.mkstemp()

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

@@ -22,17 +22,17 @@ __all__ = ["show_logs"]
 from resource_management.core.resources.system import Execute
 from resource_management.libraries.functions.format import format
 
-LAST_LINES_OUTPUT_COUNT = 40
+LAST_LINES_DEFAULT_OUTPUT_COUNT = 40
 
-def show_logs(log_dir, user):
+def show_logs(log_dir, user, lines_count=LAST_LINES_DEFAULT_OUTPUT_COUNT, mask="*"):
   """
   This should be used in 'except' block of start or stop Execute of services or of command which checks if start was fine.
   It allows to give additional debugging information without need to have access to log files.
   
   Don't forget to call "raise" after using the function or else the original exception will be masked.
   """
-  lines_count = LAST_LINES_OUTPUT_COUNT
-  Execute(format("log_files=`find {log_dir} -maxdepth 1 -type f` ; [ ! -z \"$log_files\" ] && tail -n {lines_count} $log_files"),
+  
+  Execute(format("find {log_dir} -maxdepth 1 -type f -name '{mask}' -exec echo '==> {{}} <==' \; -exec tail -n {lines_count} {{}} \;"),
           logoutput = True,
           ignore_failures = True, # if this fails should not overwrite the actual exception
           user = user, # need to have permissions to read log files

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

@@ -51,6 +51,7 @@ from resource_management.core.resources.system import Execute
 from contextlib import closing
 from resource_management.libraries.functions.stack_features import check_stack_feature
 from resource_management.libraries.functions.constants import StackFeature
+from resource_management.libraries.functions.show_logs import show_logs
 
 import ambari_simplejson as json # simplejson is much faster comparing to Python 2.6 json module and has the same functions set.
 
@@ -62,7 +63,7 @@ if OSCheck.is_windows_family():
 else:
   from resource_management.libraries.functions.tar_archive import archive_dir
 
-USAGE = """Usage: {0} <COMMAND> <JSON_CONFIG> <BASEDIR> <STROUTPUT> <LOGGING_LEVEL> <TMP_DIR>
+USAGE = """Usage: {0} <COMMAND> <JSON_CONFIG> <BASEDIR> <STROUTPUT> <LOGGING_LEVEL> <TMP_DIR> <LOG_OUT_FILES>
 
 <COMMAND> command type (INSTALL/CONFIGURE/START/STOP/SERVICE_CHECK...)
 <JSON_CONFIG> path to command json file. Ex: /var/lib/ambari-agent/data/command-2.json
@@ -70,10 +71,14 @@ USAGE = """Usage: {0} <COMMAND> <JSON_CONFIG> <BASEDIR> <STROUTPUT> <LOGGING_LEV
 <STROUTPUT> path to file with structured command output (file will be created). Ex:/tmp/my.txt
 <LOGGING_LEVEL> log level for stdout. Ex:DEBUG,INFO
 <TMP_DIR> temporary directory for executable scripts. Ex: /var/lib/ambari-agent/tmp
+<LOG_OUT_FILES> before start is done, should the service *.out files content be logged. Ex: false 
 """
 
 _PASSWORD_MAP = {"/configurations/cluster-env/hadoop.user.name":"/configurations/cluster-env/hadoop.user.password"}
 STACK_VERSION_PLACEHOLDER = "${stack_version}"
+COUNT_OF_LAST_LINES_OF_OUT_FILES_LOGGED = 100
+OUT_FILES_MASK = "*.out"
+AGENT_TASKS_LOG_FILE = "/var/log/ambari-agent/agent_tasks.log"
 
 def get_path_from_configuration(name, configuration):
   subdicts = filter(None, name.split('/'))
@@ -194,8 +199,8 @@ class Script(object):
     Parses command parameters and executes method relevant to command type
     """
     # parse arguments
-    if len(sys.argv) < 7:
-     print "Script expects at least 6 arguments"
+    if len(sys.argv) < 8:
+     print "Script expects at least 7 arguments"
      print USAGE.format(os.path.basename(sys.argv[0])) # print to stdout
      sys.exit(1)
      
@@ -206,7 +211,8 @@ class Script(object):
     self.load_structured_out()
     self.logging_level = sys.argv[5]
     Script.tmp_dir = sys.argv[6]
-
+    self.log_out_files = sys.argv[7].lower() == "true"
+    
     logging_level_str = logging._levelNames[self.logging_level]
     Logger.initialize_logger(__name__, logging_level=logging_level_str)
 
@@ -235,10 +241,39 @@ class Script(object):
       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 self.command_name == "start" and not self.is_hook():
+          self.pre_start()
+        
         method(env)
     finally:
       if self.should_expose_component_version(self.command_name):
         self.save_component_version_to_structured_out()
+        
+  def is_hook(self):
+    from resource_management.libraries.script.hook import Hook
+    return (Hook in self.__class__.__bases__)
+        
+  def get_log_folder(self):
+    return ""
+  
+  def get_user(self):
+    return ""
+        
+  def pre_start(self):
+    if self.log_out_files:
+      log_folder = self.get_log_folder()
+      user = self.get_user()
+      
+      if log_folder == "":
+        Logger.logger.warn("Log folder for current script is not defined")
+        return
+      
+      if user == "":
+        Logger.logger.warn("User for current script is not defined")
+        return
+      
+      show_logs(log_folder, user, lines_count=COUNT_OF_LAST_LINES_OF_OUT_FILES_LOGGED, mask=OUT_FILES_MASK)
 
   def choose_method_to_execute(self, command_name):
     """
@@ -616,6 +651,7 @@ class Script(object):
 
       # To remain backward compatible with older stacks, only pass upgrade_type if available.
       # TODO, remove checking the argspec for "upgrade_type" once all of the services support that optional param.
+      self.pre_start()
       if "upgrade_type" in inspect.getargspec(self.start).args:
         self.start(env, upgrade_type=upgrade_type)
       else:

ambari-server/src/main/resources/common-services/ACCUMULO/1.6.1.2.2.0/package/scripts/accumulo_script.py

@@ -169,7 +169,14 @@ class AccumuloScript(Script):
         issues.append("Configuration file %s did not pass the validation. Reason: %s" % (cf, result_issues[cf]))
       self.put_structured_out({"securityIssuesFound": ". ".join(issues)})
       self.put_structured_out({"securityState": "UNSECURED"})
+      
+  def get_log_folder(self):
+    import params
+    return params.log_dir
 
+  def get_user(self):
+    import params
+    return params.accumulo_user
 
 if __name__ == "__main__":
   AccumuloScript().fail_with_error('component unspecified')

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/package/scripts/metrics_collector.py

@@ -59,6 +59,14 @@ class AmsCollector(Script):
     import status_params
     env.set_params(status_params)
     check_service_status(name='collector')
+    
+  def get_log_folder(self):
+    import params
+    return params.ams_collector_log_dir
+  
+  def get_user(self):
+    import params
+    return params.ams_user
 
 
 @OsFamilyImpl(os_family=OsFamilyImpl.DEFAULT)

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/package/scripts/metrics_monitor.py

@@ -54,6 +54,14 @@ class AmsMonitor(Script):
     import status_params
     env.set_params(status_params)
     check_service_status(name='monitor')
+    
+  def get_log_folder(self):
+    import params
+    return params.ams_monitor_log_dir
+  
+  def get_user(self):
+    import params
+    return params.ams_user
 
 
 if __name__ == "__main__":

ambari-server/src/main/resources/common-services/ATLAS/0.1.0.2.3/package/scripts/metadata_server.py

@@ -143,6 +143,14 @@ class MetadataServer(Script):
         issues.append("Configuration file %s did not pass the validation. Reason: %s" % (cf, result_issues[cf]))
       self.put_structured_out({"securityIssuesFound": ". ".join(issues)})
       self.put_structured_out({"securityState": "UNSECURED"})
+      
+  def get_log_folder(self):
+    import params
+    return params.log_dir
+  
+  def get_user(self):
+    import params
+    return params.metadata_user
 
 if __name__ == "__main__":
   MetadataServer().execute()

ambari-server/src/main/resources/common-services/FALCON/0.5.0.2.1/package/scripts/falcon_server.py

@@ -146,6 +146,14 @@ class FalconServerLinux(FalconServer):
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
 
+  def get_log_folder(self):
+    import params
+    return params.falcon_log_dir
+  
+  def get_user(self):
+    import params
+    return params.falcon_user
+
 
 @OsFamilyImpl(os_family=OSConst.WINSRV_FAMILY)
 class FalconServerWindows(FalconServer):

ambari-server/src/main/resources/common-services/FLUME/1.4.0.2.0/package/scripts/flume_handler.py

@@ -102,6 +102,14 @@ class FlumeHandlerLinux(FlumeHandler):
     # only restore on upgrade, not downgrade
     if params.upgrade_direction == Direction.UPGRADE:
       flume_upgrade.pre_start_restore()
+      
+  def get_log_folder(self):
+    import params
+    return params.flume_log_dir
+  
+  def get_user(self):
+    import params
+    return None # means that is run from the same user as ambari is run
 
 @OsFamilyImpl(os_family=OSConst.WINSRV_FAMILY)
 class FlumeHandlerWindows(FlumeHandler):

ambari-server/src/main/resources/common-services/HBASE/0.96.0.2.0/package/scripts/hbase_master.py

@@ -144,7 +144,14 @@ class HbaseMasterDefault(HbaseMaster):
         self.put_structured_out({"securityState": "UNSECURED"})
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
-
+      
+  def get_log_folder(self):
+    import params
+    return params.log_dir
+  
+  def get_user(self):
+    import params
+    return params.hbase_user
 
 if __name__ == "__main__":
   HbaseMaster().execute()

ambari-server/src/main/resources/common-services/HBASE/0.96.0.2.0/package/scripts/hbase_regionserver.py

@@ -153,6 +153,13 @@ class HbaseRegionServerDefault(HbaseRegionServer):
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
 
+  def get_log_folder(self):
+    import params
+    return params.log_dir
+  
+  def get_user(self):
+    import params
+    return params.hbase_user
 
 if __name__ == "__main__":
   HbaseRegionServer().execute()

ambari-server/src/main/resources/common-services/HBASE/0.96.0.2.0/package/scripts/phoenix_queryserver.py

@@ -75,6 +75,14 @@ class PhoenixQueryServer(Script):
 
   def security_status(self, env):
     self.put_structured_out({"securityState": "UNSECURED"})
+    
+  def get_log_folder(self):
+    import params
+    return params.log_dir
+  
+  def get_user(self):
+    import params
+    return params.hbase_user
 
 if __name__ == "__main__":
   PhoenixQueryServer().execute()

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

@@ -155,6 +155,14 @@ class DataNodeDefault(DataNode):
         self.put_structured_out({"securityState": "UNSECURED"})
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
+      
+  def get_log_folder(self):
+    import params
+    return params.hdfs_log_dir
+  
+  def get_user(self):
+    import params
+    return params.hdfs_user
 
 @OsFamilyImpl(os_family=OSConst.WINSRV_FAMILY)
 class DataNodeWindows(DataNode):

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

@@ -159,6 +159,14 @@ class JournalNodeDefault(JournalNode):
         self.put_structured_out({"securityState": "UNSECURED"})
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
+      
+  def get_log_folder(self):
+    import params
+    return params.hdfs_log_dir
+  
+  def get_user(self):
+    import params
+    return params.hdfs_user
 
 @OsFamilyImpl(os_family=OSConst.WINSRV_FAMILY)
 class JournalNodeWindows(JournalNode):

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

@@ -347,6 +347,14 @@ class NameNodeDefault(NameNode):
       File(ccache_file_path,
            action = "delete",
       )
+      
+  def get_log_folder(self):
+    import params
+    return params.hdfs_log_dir
+  
+  def get_user(self):
+    import params
+    return params.hdfs_user
 
 @OsFamilyImpl(os_family=OSConst.WINSRV_FAMILY)
 class NameNodeWindows(NameNode):

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

@@ -134,6 +134,14 @@ class NFSGateway(Script):
         self.put_structured_out({"securityState": "UNSECURED"})
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
+      
+  def get_log_folder(self):
+    import params
+    return params.hdfs_log_dir
+  
+  def get_user(self):
+    import params
+    return params.hdfs_user
 
 if __name__ == "__main__":
   NFSGateway().execute()

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

@@ -135,6 +135,14 @@ class SNameNodeDefault(SNameNode):
         self.put_structured_out({"securityState": "UNSECURED"})
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
+      
+  def get_log_folder(self):
+    import params
+    return params.hdfs_log_dir
+  
+  def get_user(self):
+    import params
+    return params.hdfs_user
 
 @OsFamilyImpl(os_family=OSConst.WINSRV_FAMILY)
 class SNameNodeWindows(SNameNode):

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

@@ -133,6 +133,14 @@ class ZkfcSlaveDefault(ZkfcSlave):
         self.put_structured_out({"securityState": "UNSECURED"})
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
+      
+  def get_log_folder(self):
+    import params
+    return params.hdfs_log_dir
+  
+  def get_user(self):
+    import params
+    return params.hdfs_user
 
 def initialize_ha_zookeeper(params):
   try:

ambari-server/src/main/resources/common-services/HIVE/0.12.0.2.0/package/scripts/hive_metastore.py

@@ -240,6 +240,14 @@ class HiveMetastoreDefault(HiveMetastore):
 
     command = format("{binary} -dbType {hive_metastore_db_type} -upgradeSchema")
     Execute(command, user=params.hive_user, tries=1, environment=env_dict, logoutput=True)
+    
+  def get_log_folder(self):
+    import params
+    return params.hive_log_dir
+
+  def get_user(self):
+    import params
+    return params.hive_user
 
 
 if __name__ == "__main__":

ambari-server/src/main/resources/common-services/HIVE/0.12.0.2.0/package/scripts/hive_server.py

@@ -199,6 +199,13 @@ class HiveServerDefault(HiveServer):
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
 
+  def get_log_folder(self):
+    import params
+    return params.hive_log_dir
+  
+  def get_user(self):
+    import params
+    return params.hive_user
 
 if __name__ == "__main__":
   HiveServer().execute()

ambari-server/src/main/resources/common-services/HIVE/0.12.0.2.0/package/scripts/hive_server_interactive.py

@@ -378,6 +378,14 @@ class HiveServerInteractiveDefault(HiveServerInteractive):
         Logger.info("App '{0}' did not come up after a wait of {1} seconds".format(llap_app_name,
                                                                                           time.time() - curr_time))
         return False
+      
+    def get_log_folder(self):
+      import params
+      return params.hive_log_dir
+    
+    def get_user(self):
+      import params
+      return params.hive_user
 
 @OsFamilyImpl(os_family=OSConst.WINSRV_FAMILY)
 class HiveServerInteractiveWindows(HiveServerInteractive):

ambari-server/src/main/resources/common-services/HIVE/0.12.0.2.0/package/scripts/webhcat_server.py

@@ -152,6 +152,13 @@ class WebHCatServerDefault(WebHCatServer):
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
 
+  def get_log_folder(self):
+    import params
+    return params.hcat_log_dir
+  
+  def get_user(self):
+    import params
+    return params.webhcat_user
 
 if __name__ == "__main__":
   WebHCatServer().execute()

ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/kafka_broker.py

@@ -116,6 +116,14 @@ class KafkaBroker(Script):
     import status_params
     env.set_params(status_params)
     check_process_status(status_params.kafka_pid_file)
+    
+  def get_log_folder(self):
+    import params
+    return params.kafka_log_dir
+  
+  def get_user(self):
+    import params
+    return params.kafka_user
 
 if __name__ == "__main__":
   KafkaBroker().execute()

ambari-server/src/main/resources/common-services/KNOX/0.5.0.2.2/package/scripts/knox_gateway.py

@@ -278,6 +278,14 @@ class KnoxGatewayDefault(KnoxGateway):
         self.put_structured_out({"securityState": "UNSECURED"})
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
+      
+  def get_log_folder(self):
+    import params
+    return params.knox_logs_dir
+  
+  def get_user(self):
+    import params
+    return params.knox_user
 
 
 if __name__ == "__main__":

ambari-server/src/main/resources/common-services/OOZIE/4.0.0.2.0/package/scripts/oozie_server.py

@@ -200,6 +200,14 @@ class OozieServerDefault(OozieServer):
 
     OozieUpgrade.restore_configuration()
     OozieUpgrade.prepare_libext_directory()
+    
+  def get_log_folder(self):
+    import params
+    return params.oozie_log_dir
+  
+  def get_user(self):
+    import params
+    return params.oozie_user
 
 
 @OsFamilyImpl(os_family=OSConst.WINSRV_FAMILY)

ambari-server/src/main/resources/common-services/RANGER/0.4.0/package/scripts/ranger_admin.py

@@ -151,6 +151,14 @@ class RangerAdmin(Script):
       from setup_ranger_xml import setup_java_patch
       setup_java_patch(stack_version=stack_version)
 
+  def get_log_folder(self):
+    import params
+    return params.admin_log_dir
+  
+  def get_user(self):
+    import params
+    return params.unix_user
+
 if __name__ == "__main__":
   RangerAdmin().execute()
 

ambari-server/src/main/resources/common-services/RANGER/0.4.0/package/scripts/ranger_tagsync.py

@@ -75,6 +75,13 @@ class RangerTagsync(Script):
   def get_component_name(self):
     return "ranger-tagsync"
 
+  def get_log_folder(self):
+    import params
+    return params.tagsync_log_dir
+  
+  def get_user(self):
+    import params
+    return params.unix_user
 
 if __name__ == "__main__":
   RangerTagsync().execute()

ambari-server/src/main/resources/common-services/RANGER/0.4.0/package/scripts/ranger_usersync.py

@@ -73,6 +73,13 @@ class RangerUsersync(Script):
   def get_component_name(self):
     return "ranger-usersync"
 
+  def get_log_folder(self):
+    import params
+    return params.usersync_log_dir
+  
+  def get_user(self):
+    import params
+    return params.unix_user
 
 if __name__ == "__main__":
   RangerUsersync().execute()

ambari-server/src/main/resources/common-services/RANGER_KMS/0.5.0.2.3/package/scripts/kms_server.py

@@ -91,6 +91,14 @@ class KmsServer(Script):
     stack_version = upgrade_stack[1]
     Logger.info(format('Setting Ranger KMS database schema, using version {stack_version}'))
     setup_kms_db(stack_version=stack_version)
+    
+  def get_log_folder(self):
+    import params
+    return params.kms_log_dir
+  
+  def get_user(self):
+    import params
+    return params.kms_user
 
 if __name__ == "__main__":
   KmsServer().execute()

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

@@ -90,6 +90,14 @@ class JobHistoryServer(Script):
           host_sys_prepped=params.host_sys_prepped)
         if resource_created:
           params.HdfsResource(None, action="execute")
+          
+  def get_log_folder(self):
+    import params
+    return params.spark_log_dir
+  
+  def get_user(self):
+    import params
+    return params.spark_user
 
 if __name__ == "__main__":
   JobHistoryServer().execute()

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

@@ -74,6 +74,14 @@ class SparkThriftServer(Script):
       Logger.info("Executing Spark Thrift Server Stack Upgrade pre-restart")
       conf_select.select(params.stack_name, "spark", params.version)
       stack_select.select("spark-thriftserver", params.version)
+      
+  def get_log_folder(self):
+    import params
+    return params.spark_log_dir
+  
+  def get_user(self):
+    import params
+    return params.hive_user
 
 if __name__ == "__main__":
   SparkThriftServer().execute()

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/drpc_server.py

@@ -126,6 +126,14 @@ class DrpcServer(Script):
         self.put_structured_out({"securityStateErrorInfo": str(e)})
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
+      
+  def get_log_folder(self):
+    import params
+    return params.log_dir
+  
+  def get_user(self):
+    import params
+    return params.storm_user
 
 if __name__ == "__main__":
   DrpcServer().execute()

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/nimbus.py

@@ -127,6 +127,13 @@ class NimbusDefault(Nimbus):
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
 
+  def get_log_folder(self):
+    import params
+    return params.log_dir
+  
+  def get_user(self):
+    import params
+    return params.storm_user
 
 @OsFamilyImpl(os_family=OSConst.WINSRV_FAMILY)
 class NimbusWindows(Nimbus):

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/nimbus_prod.py

@@ -68,6 +68,14 @@ class Nimbus(Script):
 
   def status(self, env):
     supervisord_check_status("nimbus")
+    
+  def get_log_folder(self):
+    import params
+    return params.log_dir
+  
+  def get_user(self):
+    import params
+    return params.storm_user
 
 if __name__ == "__main__":
   Nimbus().execute()

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/rest_api.py

@@ -69,5 +69,13 @@ class StormRestApi(Script):
     env.set_params(status_params)
     check_process_status(status_params.pid_rest_api)
 
+  def get_log_folder(self):
+    import params
+    return params.log_dir
+  
+  def get_user(self):
+    import params
+    return params.storm_user
+  
 if __name__ == "__main__":
   StormRestApi().execute()

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/supervisor.py

@@ -100,6 +100,13 @@ class SupervisorDefault(Supervisor):
     env.set_params(status_params)
     check_process_status(status_params.pid_supervisor)
 
+  def get_log_folder(self):
+    import params
+    return params.log_dir
+  
+  def get_user(self):
+    import params
+    return params.storm_user
 
 if __name__ == "__main__":
   Supervisor().execute()

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/supervisor_prod.py

@@ -71,6 +71,14 @@ class Supervisor(Script):
 
   def status(self, env):
     supervisord_check_status("supervisor")
+    
+  def get_log_folder(self):
+    import params
+    return params.log_dir
+  
+  def get_user(self):
+    import params
+    return params.storm_user
 
 if __name__ == "__main__":
   Supervisor().execute()

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/ui_server.py

@@ -167,6 +167,14 @@ class UiServerDefault(UiServer):
         self.put_structured_out({"securityStateErrorInfo": str(e)})
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
+      
+  def get_log_folder(self):
+    import params
+    return params.log_dir
+  
+  def get_user(self):
+    import params
+    return params.storm_user
 
 if __name__ == "__main__":
   UiServer().execute()

ambari-server/src/main/resources/common-services/YARN/2.1.0.2.0/package/scripts/application_timeline_server.py

@@ -143,6 +143,13 @@ class ApplicationTimelineServerDefault(ApplicationTimelineServer):
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
 
+  def get_log_folder(self):
+    import params
+    return params.yarn_log_dir
+  
+  def get_user(self):
+    import params
+    return params.yarn_user
 
 if __name__ == "__main__":
   ApplicationTimelineServer().execute()

ambari-server/src/main/resources/common-services/YARN/2.1.0.2.0/package/scripts/historyserver.py

@@ -178,6 +178,13 @@ class HistoryServerDefault(HistoryServer):
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
 
+  def get_log_folder(self):
+    import params
+    return params.mapred_log_dir
+
+  def get_user(self):
+    import params
+    return params.mapred_user
 
 if __name__ == "__main__":
   HistoryServer().execute()

ambari-server/src/main/resources/common-services/YARN/2.1.0.2.0/package/scripts/nodemanager.py

@@ -159,6 +159,13 @@ class NodemanagerDefault(Nodemanager):
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
 
+  def get_log_folder(self):
+    import params
+    return params.yarn_log_dir
+  
+  def get_user(self):
+    import params
+    return params.yarn_user
 
 if __name__ == "__main__":
   Nodemanager().execute()

ambari-server/src/main/resources/common-services/YARN/2.1.0.2.0/package/scripts/resourcemanager.py

@@ -275,5 +275,13 @@ class ResourcemanagerDefault(Resourcemanager):
       else:
         Logger.info("DFS directory '" + dir_path + "' exists.")
 
+  def get_log_folder(self):
+    import params
+    return params.yarn_log_dir
+  
+  def get_user(self):
+    import params
+    return params.yarn_user
+  
 if __name__ == "__main__":
   Resourcemanager().execute()

ambari-server/src/main/resources/common-services/ZOOKEEPER/3.4.5.2.0/package/scripts/zookeeper_server.py

@@ -161,6 +161,14 @@ class ZookeeperServerLinux(ZookeeperServer):
         self.put_structured_out({"securityStateErrorInfo": str(e)})
     else:
       self.put_structured_out({"securityState": "UNSECURED"})
+      
+  def get_log_folder(self):
+    import params
+    return params.zk_log_dir
+  
+  def get_user(self):
+    import params
+    return params.zk_user
 
 
 @OsFamilyImpl(os_family=OSConst.WINSRV_FAMILY)

ambari-server/src/test/python/stacks/2.0.6/HIVE/test_webhcat_server.py

@@ -106,11 +106,11 @@ class TestWebHCatServer(RMFTestCase):
                        target = RMFTestCase.TARGET_COMMON_SERVICES
                        )
 
-    self.assertResourceCalled('Execute', 'log_files=`find /var/log/webhcat -maxdepth 1 -type f` ; [ ! -z "$log_files" ] && tail -n 40 $log_files',
-                              logoutput = True,
-                              ignore_failures = True,
-                              user = 'hcat'
-                             )
+    self.assertResourceCalled('Execute', "find /var/log/webhcat -maxdepth 1 -type f -name '*' -exec echo '==> {} <==' \\; -exec tail -n 40 {} \\;",
+        logoutput = True,
+        ignore_failures = True,
+        user = 'hcat',
+    )
 
     self.assertResourceCalled('Execute', 'ambari-sudo.sh kill -9 `ambari-sudo.sh su hcat -l -s /bin/bash -c \'[RMF_EXPORT_PLACEHOLDER]cat /var/run/webhcat/webhcat.pid\'`',
                               not_if = "! (ls /var/run/webhcat/webhcat.pid >/dev/null 2>&1 && ps -p `ambari-sudo.sh su hcat -l -s /bin/bash -c '[RMF_EXPORT_PLACEHOLDER]cat /var/run/webhcat/webhcat.pid'` >/dev/null 2>&1) || ( sleep 10 && ! (ls /var/run/webhcat/webhcat.pid >/dev/null 2>&1 && ps -p `ambari-sudo.sh su hcat -l -s /bin/bash -c '[RMF_EXPORT_PLACEHOLDER]cat /var/run/webhcat/webhcat.pid'` >/dev/null 2>&1) )"
@@ -180,11 +180,11 @@ class TestWebHCatServer(RMFTestCase):
                        target = RMFTestCase.TARGET_COMMON_SERVICES
                        )
 
-    self.assertResourceCalled('Execute', 'log_files=`find /var/log/webhcat -maxdepth 1 -type f` ; [ ! -z "$log_files" ] && tail -n 40 $log_files',
-                              logoutput = True,
-                              ignore_failures = True,
-                              user = 'hcat'
-                              )
+    self.assertResourceCalled('Execute', "find /var/log/webhcat -maxdepth 1 -type f -name '*' -exec echo '==> {} <==' \\; -exec tail -n 40 {} \\;",
+        logoutput = True,
+        ignore_failures = True,
+        user = 'hcat',
+    )
     
     self.assertResourceCalled('Execute', 'ambari-sudo.sh kill -9 `ambari-sudo.sh su hcat -l -s /bin/bash -c \'[RMF_EXPORT_PLACEHOLDER]cat /var/run/webhcat/webhcat.pid\'`',
                               not_if = "! (ls /var/run/webhcat/webhcat.pid >/dev/null 2>&1 && ps -p `ambari-sudo.sh su hcat -l -s /bin/bash -c '[RMF_EXPORT_PLACEHOLDER]cat /var/run/webhcat/webhcat.pid'` >/dev/null 2>&1) || ( sleep 10 && ! (ls /var/run/webhcat/webhcat.pid >/dev/null 2>&1 && ps -p `ambari-sudo.sh su hcat -l -s /bin/bash -c '[RMF_EXPORT_PLACEHOLDER]cat /var/run/webhcat/webhcat.pid'` >/dev/null 2>&1) )"

ambari-server/src/test/python/stacks/utils/RMFTestCase.py

@@ -71,7 +71,8 @@ class RMFTestCase(TestCase):
                     target=TARGET_STACKS,
                     mocks_dict={},
                     try_install=False,
-                    command_args=[]):
+                    command_args=[],
+                    log_out_files=False):
     norm_path = os.path.normpath(path)
     src_dir = RMFTestCase.get_src_folder()
     if target == self.TARGET_STACKS:
@@ -115,6 +116,7 @@ class RMFTestCase(TestCase):
       with patch.object(platform, 'linux_distribution', return_value=os_type):
         script_module = imp.load_source(classname, script_path)
         script_class_inst = RMFTestCase._get_attr(script_module, classname)()
+        script_class_inst.log_out_files = log_out_files
         method = RMFTestCase._get_attr(script_class_inst, command)
     except IOError, err:
       raise RuntimeError("Cannot load class %s from %s: %s" % (classname, norm_path, err.message))