PNDA-4237: Reporting failed YARN application submission

Author: dharaneeshvrd

api/src/main/resources/application_summary.py

@@ -1,4 +1,5 @@
 import json
+import commands
 import multiprocessing
 import time
 import logging
@@ -368,43 +369,85 @@ def check_in_yarn(job_name):
         logging.debug(str(error_message))
     return run_app_info
 
-def spark_application(job_name):
+# pylint: disable=C0103
+
+def check_in_service_log(namespace, application, component_name):
+    '''
+    Check in service log in case of application failed to submit to YARN
+    '''
+    service_name = '%s-%s-%s' % (namespace, application, component_name)
+    (command, message, more_detail) = ('', '', '')
+    command = 'sudo journalctl -u %s.service' % service_name
+    out = commands.getoutput('%s -n 50' % command).split('\n')
+    more_detail = 'More details: execute "journalctl -u %s"' % service_name
+    for line in out:
+        if 'Exception:' in line:
+            message = '%s%s' % (line.split('Exception:')[0].split(' ')[-1], 'Exception')
+            break
+    if message == '':
+        message = '%s' % (more_detail)
+    else:
+        message = '%s. %s' % (message, more_detail)
+    return 'FAILED_TO_SUBMIT_TO_YARN', message
+
+def spark_yarn_handler(yarn_data):
+    '''
+    Handling Spark YARN data
+    '''
+    information = ''
+    aggregate_status = ''
+    yarnid = yarn_data['id']
+    if yarn_data['state'] == 'SUBMITTED' or yarn_data['state'] == 'ACCEPTED':
+        aggregate_status = yarn_data['state']
+        message = yarn_data['diagnostics'].split('Details :')[0].strip()
+        information = message
+    elif yarn_data['state'] == 'RUNNING':
+        spark_data = spark_job_handler(yarn_data['id'])
+        if spark_data['state'] == 'OK':
+            aggregate_status = 'RUNNING'
+        else:
+            aggregate_status = 'RUNNING_WITH_ERRORS'
+        information = spark_data['information']
+    elif yarn_data['finalStatus'] == 'SUCCEEDED':
+        aggregate_status = '%s_%s' % (yarn_data['state'], yarn_data['finalStatus'])
+    elif yarn_data['state'] == 'FINISHED' and (yarn_data['finalStatus'] == 'FAILED' or yarn_data['finalStatus'] == 'KILLED'):
+        aggregate_status = '%s_%s' % (yarn_data['state'], yarn_data['finalStatus'])
+        information = yarn_data['diagnostics']
+    elif yarn_data['finalStatus'] == 'FAILED' or yarn_data['finalStatus'] == 'KILLED':
+        aggregate_status = yarn_data['finalStatus']
+        information = yarn_data['diagnostics']
+    else:
+        aggregate_status = 'NOT_FOUND'
+        message = yarn_data.get('RemoteException', {'message': ['']}).\
+        get('message').split(':')
+        message[0] = ''
+        information = ''.join(message).strip()
+    return aggregate_status, yarnid, information
+
+def spark_application(job_name, application, component_name):
     """
     Handling SPARK Application
     """
     ret = {}
-    yarnid = ''
-    information = ''
+    check_in_service = False
+    (aggregate_status, yarnid, information) = ('', '', '')
+    status, timestamp = _HBASE.get_status_with_timestamp(application)
     yarn_data = check_in_yarn(job_name)
-    if yarn_data != None:
-        yarnid = yarn_data['id']
-        if yarn_data['state'] == 'SUBMITTED' or yarn_data['state'] == 'ACCEPTED':
-            aggregate_status = yarn_data['state']
-            message = yarn_data['diagnostics'].split('Details :')[0].strip()
-            information = message
-        elif yarn_data['state'] == 'RUNNING':
-            spark_data = spark_job_handler(yarn_data['id'])
-            if spark_data['state'] == 'OK':
-                aggregate_status = 'RUNNING'
-            else:
-                aggregate_status = 'RUNNING_WITH_ERRORS'
-            information = spark_data['information']
-        elif yarn_data['finalStatus'] == 'SUCCEEDED':
-            aggregate_status = '%s_%s' % (yarn_data['state'], yarn_data['finalStatus'])
-        elif yarn_data['state'] == 'FINISHED' and (yarn_data['finalStatus'] == 'FAILED' or yarn_data['finalStatus'] == 'KILLED'):
-            aggregate_status = '%s_%s' % (yarn_data['state'], yarn_data['finalStatus'])
-            information = yarn_data['diagnostics']
-        elif yarn_data['finalStatus'] == 'FAILED' or yarn_data['finalStatus'] == 'KILLED':
-            aggregate_status = yarn_data['finalStatus']
-            information = yarn_data['diagnostics']
+    if status == 'CREATED':
+        if yarn_data != None:
+            aggregate_status, yarnid, information = spark_yarn_handler(yarn_data)
         else:
-            aggregate_status = 'NOT_FOUND'
-            message = yarn_data.get('RemoteException', {'message': ['']}).\
-            get('message').split(':')
-            message[0] = ''
-            information = ''.join(message).strip()
+            aggregate_status = ApplicationState.CREATED
     else:
-        aggregate_status = ApplicationState.CREATED
+        if yarn_data != None:
+            if timestamp < yarn_data['startedTime']:
+                aggregate_status, yarnid, information = spark_yarn_handler(yarn_data)
+            else:
+                check_in_service = True
+        else:
+            check_in_service = True
+    if check_in_service:
+        aggregate_status, information = check_in_service_log(CONFIG['environment']['namespace'], application, component_name)
     ret = {
         'aggregate_status': aggregate_status,
         'yarnId': yarnid,
@@ -431,7 +474,8 @@ def get_json(component_list, queue_obj):
                     (component[application][component_name]['job_handle'])})
                 if 'sparkStreaming' in component_name:
                     ret[application].update({component_name: spark_application\
-                    (component[application][component_name]['component_job_name'])})
+                    (component[application][component_name]['component_job_name'], application,\
+                     component[application][component_name]['component_name'])})
     queue_obj.put([{process_name: ret}])
     logging.info('%s %s', 'Finished', process_name)
 

api/src/main/resources/application_summary_registrar.py

@@ -82,6 +82,18 @@ def get_dm_data(self, key):
             connection.close()
         return data
 
+    def get_status_with_timestamp(self, key):
+        try:
+            connection = happybase.Connection(self._hbase_host)
+            table = connection.table("platform_applications")
+            row = table.row(key, columns=['cf:status'], include_timestamp=True)
+            status, timestamp = row['cf:status']
+        except TTransportException as error_message:
+            logging.error(str(error_message))
+        finally:
+            connection.close()
+        return status, timestamp
+
     def get_summary_data(self, application):
         record = {application: {}}
         dm_data = self.get_dm_data(application)

api/src/main/resources/test_application_summary.py

@@ -17,43 +17,42 @@ def test_process_application_data(self):
         {'aggregate_status': 'RUNNING_WITH_ERRORS', 'yarnId': '1234'}})
         self.assertEqual(result, 'RUNNING_WITH_ERRORS')
 
-    @patch('application_summary.check_in_yarn')
     @patch('application_summary.spark_job_handler')
-    def test_spark_application(self, spark_job_patch, yarn_check_patch):
+    def test_spark_yarn_handler(self, spark_job_patch):
         """
         Tetsing Spark application
         """
         #Spark application in case of Killed
-        yarn_check_patch.return_value = {'id': 'application_1234', 'state': 'KILLED', 'finalStatus': 'KILLED', \
+        input_data = {'id': 'application_1234', 'state': 'KILLED', 'finalStatus': 'KILLED', \
         'diagnostics': 'Killed'}
-        result = application_summary.spark_application('app1-example-job')
-        self.assertEqual(result['aggregate_status'], '%s' % ('KILLED'))
+        result = application_summary.spark_yarn_handler(input_data)
+        self.assertEqual(result[0], '%s' % ('KILLED'))
 
         #Spark application in case of Failed
-        yarn_check_patch.return_value = {'id': 'application_1234', 'state': 'FINISHED', 'finalStatus': 'FAILED', \
+        input_data = {'id': 'application_1234', 'state': 'FINISHED', 'finalStatus': 'FAILED', \
         'diagnostics': 'Failed'}
-        result = application_summary.spark_application('app1-example-job')
-        self.assertEqual(result['aggregate_status'], '%s' % ('FINISHED_FAILED'))
+        result = application_summary.spark_yarn_handler(input_data)
+        self.assertEqual(result[0], '%s' % ('FINISHED_FAILED'))
 
         #Spark application in case of Running with No errors
-        yarn_check_patch.return_value = {'id': 'application_1234', 'state': 'RUNNING', \
+        input_data = {'id': 'application_1234', 'state': 'RUNNING', \
         'finalStatus': 'UNDEFINED'}
         spark_job_patch.return_value = {'state': 'OK', 'information': 'job_stage data'}
-        result = application_summary.spark_application('app1-example-job')
-        self.assertEqual(result['aggregate_status'], '%s' % ('RUNNING'))
+        result = application_summary.spark_yarn_handler(input_data)
+        self.assertEqual(result[0], '%s' % ('RUNNING'))
 
         #Spark application in case of Running with errors
-        yarn_check_patch.return_value = {'id': 'application_1234', 'state': 'RUNNING', \
+        input_data = {'id': 'application_1234', 'state': 'RUNNING', \
         'finalStatus': 'UNDEFINED'}
         spark_job_patch.return_value = {'state': 'ERROR', 'information': 'job_stage data'}
-        result = application_summary.spark_application('app1-example-job')
-        self.assertEqual(result['aggregate_status'], '%s' % ('RUNNING_WITH_ERRORS'))
+        result = application_summary.spark_yarn_handler(input_data)
+        self.assertEqual(result[0], '%s' % ('RUNNING_WITH_ERRORS'))
 
         #Spark application in other states than above states
-        yarn_check_patch.return_value = {'id': 'application_1234', 'state': 'ACCEPTED', \
+        input_data = {'id': 'application_1234', 'state': 'ACCEPTED', \
         'finalStatus': 'UNDEFINED', 'diagnostics': 'Accepted'}
-        result = application_summary.spark_application('app1-example-job')
-        self.assertEqual(result['aggregate_status'], '%s' % ('ACCEPTED'))
+        result = application_summary.spark_yarn_handler(input_data)
+        self.assertEqual(result[0], '%s' % ('ACCEPTED'))
 
     @patch('requests.get')
     def test_check_in_yarn(self, mock_req):
@@ -473,3 +472,14 @@ def test_yarn_info(self, yarn_mock_req):
             'type': 'UNKNOWN',
             'information': 'app with id application_123 not found'
         })
+
+# pylint: disable=C0301
+
+    @patch('platform.dist')
+    @patch('commands.getoutput')
+    def test_check_in_service_log(self, cmd_patch, distro_patch):
+        distro_patch.return_value = ['redhat']
+        cmd_patch.return_value = 'Mar 22 03:39:32 rhel-cdh-hadoop-edge spark-submit[2475]: 18/03/22 03:39:32 INFO yarn.Client: Uploading resource file:/opt/platform_app/s1/example/dataplatform-raw.avsc -> hdfs://rhel-cdh-hadoop-mgr-1:8020/user/pnda/.sparkStaging/application_1521689436801_0013/dataplatform-raw.avsc\nMar 22 03:39:32 rhel-cdh-hadoop-edge spark-submit[2475]: 18/03/22 03:39:32 INFO yarn.Client: Uploading resource file:/opt/platform_app/s1/example/avro-1.8.1-py2.7.egg -> hdfs://rhel-cdh-hadoop-mgr-1:8020/user/pnda/.sparkStaging/application_1521689436801_0013/avro-1.8.1-py2.7.egg\nMar 22 03:39:32 rhel-cdh-hadoop-edge spark-submit[2475]: 18/03/22 03:39:32 INFO yarn.Client: Deleting staging directory .sparkStaging/application_1521689436801_0013\nMar 22 03:39:32 rhel-cdh-hadoop-edge spark-submit[2475]: Exception in thread "main" java.io.FileNotFoundException: File file:/opt/platform_app/s1/example/avro-1.8.1-py2.7.egg does not exist\nMar 22 03:38:01 rhel-cdh-hadoop-edge spark-submit[31045]: at org.apache.hadoop.fs.RawLocalFileSystem.deprecatedGetFileStatus(RawLocalFileSystem.java:598)\nMar 22 03:38:01 rhel-cdh-hadoop-edge spark-submit[31045]: at org.apache.hadoop.fs.RawLocalFileSystem.getFileLinkStatusInternal(RawLocalFileSystem.java:811)\nMar 22 03:38:01 rhel-cdh-hadoop-edge spark-submit[31045]: at org.apache.hadoop.fs.RawLocalFileSystem.getFileStatus(RawLocalFileSystem.java:588)\nMar 22 03:38:01 rhel-cdh-hadoop-edge spark-submit[31045]: at org.apache.hadoop.fs.FilterFileSystem.getFileStatus(FilterFileSystem.java:425)\nMar 22 03:38:01 rhel-cdh-hadoop-edge spark-submit[31045]: at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:340)Mar 22 03:38:01 rhel-cdh-hadoop-edge spark-submit[31045]: at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:292)'
+        result = application_summary.check_in_service_log('platform_app', 'application_123', 'example')
+        self.assertEqual(result[0], 'FAILED_TO_SUBMIT_TO_YARN')
+        self.assertEqual(result[1], 'java.io.FileNotFoundException. More details: execute "journalctl -u platform_app-application_123-example"')