Refactor application detail API

jeclarke · jeclarke · commit 50e7d98e48cf · 2016-11-24T15:54:30.000Z
Add yarn status &amp; return the status for most recent app instance, not
just a running one.

Use the RM REST API instead of the CLI to get this information.

PNDA-2464
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,6 +4,7 @@ All notable changes to this project will be documented in this file.
 ## [Unreleased]
 ### Changed
 - Externalized build logic from Jenkins to shell script so it can be reused
+- Refactored the information returned by the Application Detail API to include the YARN application state and also to return information for jobs that have ended. Made the implementation more performant by using the YARN Resource Manager REST API instead of the CLI.
 
 ## [0.2.0] 2016-10-21
 ### Added
diff --git a/README.md b/README.md
@@ -255,12 +255,19 @@ Response Codes:
 500 - Server Error
 
 {
-        "status": "STARTED",
-        "name": "spark-batch-example-app-instance",
-        "yarn-ids": [
-            {"component":"example", "type":"oozie", "yarn-id":"application_1455877292606_0404"}
-        ]
+        "yarn_applications": {
+		    "oozie-example": {
+			    "type": "oozie",
+				"yarn-id": "application_1479988623709_0015",
+				"component": "example",
+				"yarn-start-time": 1479992520527,
+				"yarn-state": "FINISHED"
+			}
+		},
+		"status": "STARTED",
+		"name": "spark-batch-example-app-instance"
 }
+
 ````
 
 ### Start _application_
diff --git a/api/src/main/resources/application_creator.py b/api/src/main/resources/application_creator.py
@@ -134,11 +134,11 @@ def get_application_runtime_details(self, application_name, application_create_d
         logging.debug("get_application_runtime_details: %s %s", application_name, application_create_data)
 
         details = {}
-        details['yarn_ids'] = []
+        details['yarn_applications'] = {}
         for component_type, component_create_data in application_create_data.iteritems():
             creator = self._load_creator(component_type)
             type_details = creator.get_component_runtime_details(component_create_data)
-            details['yarn_ids'].extend(type_details['yarn_ids'])
+            details['yarn_applications'].update(type_details['yarn_applications'])
         return details
 
     def _load_creator(self, component_type):
diff --git a/api/src/main/resources/deployer_utils.py b/api/src/main/resources/deployer_utils.py
@@ -88,10 +88,13 @@ def fill_hadoop_env(env):
         elif service.type == "YARN":
             for role in service.get_all_roles():
                 if role.type == "RESOURCEMANAGER":
-                    env['yarn_resource_manager_host'] = '%s' % api.get_host(
-                        role.hostRef.hostId).hostname
-                    env['yarn_resource_manager_port'] = '8088'
-                    env['yarn_resource_manager_mr_port'] = '8032'
+                    if 'yarn_resource_manager_host' in env:
+                        rm_instance = '_backup'
+                    else:
+                        rm_instance = ''
+                    env['yarn_resource_manager_host%s' % rm_instance] = '%s' % api.get_host(role.hostRef.hostId).hostname
+                    env['yarn_resource_manager_port%s' % rm_instance] = '8088'
+                    env['yarn_resource_manager_mr_port%s' % rm_instance] = '8032'
                 if role.type == "NODEMANAGER":
                     if 'yarn_node_managers' in env:
                         env['yarn_node_managers'] = '%s,%s' % (
diff --git a/api/src/main/resources/plugins/base_creator.py b/api/src/main/resources/plugins/base_creator.py
@@ -36,7 +36,7 @@
 import json
 import string
 import collections
-import subprocess
+import requests
 import hbase_descriptor
 import opentsdb_descriptor
 from deployer_utils import HDFS
@@ -59,6 +59,18 @@ def __init__(self, config, environment, namespace):
                                  environment['webhdfs_port'],
                                  'hdfs')
 
+        if 'yarn_resource_manager_host' in environment:
+            self._yarn_resource_manager = "%s:%s" % (environment['yarn_resource_manager_host'],
+                                                     environment['yarn_resource_manager_port'])
+        else:
+            self._yarn_resource_manager = None
+
+        if 'yarn_resource_manager_host_backup' in environment:
+            self._yarn_resource_manager_backup = "%s:%s" % (environment['yarn_resource_manager_host_backup'],
+                                                            environment['yarn_resource_manager_port_backup'])
+        else:
+            self._yarn_resource_manager_backup = None
+
     def validate_component(self, components):
         '''
         Validates components of the package of given component type
@@ -126,6 +138,12 @@ def stop_component(self, application_name, stop_data):
         '''
         pass
 
+    def get_component_type(self):
+        '''
+        returns a name for the component type
+        '''
+        pass
+
     def _instantiate_properties(self, application_name, component, property_overrides):
         logging.debug(
             "_instantiate_properties %s %s",
@@ -257,24 +275,55 @@ def validate_components(self, components):
 
     def get_component_runtime_details(self, create_data):
         logging.debug("get_component_runtime_details: %s", create_data)
+
         details = {}
-        yarn_ids = []
-        details['yarn_ids'] = yarn_ids
-        for single_component_data in create_data:
-            yarn_id = self.get_yarn_id(single_component_data['component_job_name'])
-            if yarn_id is not None:
-                yarn_ids.append({"component": single_component_data['component_name'],
-                                 "type": self.get_component_type(),
-                                 "yarn-id": yarn_id})
+        yarn_applications = {}
+        details['yarn_applications'] = yarn_applications
+
+        all_yarn_applications = self._get_yarn_applications()
+        if all_yarn_applications is not None:
+            for single_component_data in create_data:
+                app_info = self._find_yarn_app_info(all_yarn_applications, single_component_data['component_job_name'])
+                if app_info is not None:
+                    job_key = '%s-%s' % (self.get_component_type(), single_component_data['component_name'])
+                    yarn_applications[job_key] = {"component": single_component_data['component_name'],
+                                                  "type": self.get_component_type(),
+                                                  "yarn-id": app_info['id'],
+                                                  "yarn-start-time": app_info['startedTime'],
+                                                  "yarn-state": app_info['state']}
+        else:
+            logging.error('Failed to query application list from any resource manager')
+
         return details
 
-    def get_yarn_id(self, job_name):
-        out = subprocess.check_output(['yarn', 'application', '-list'])
-        for line in out.splitlines():
-            fields = line.split('\t')
-            if len(fields) >= 6:
-                logging.debug(line)
-                app = fields[1].strip()
-                if app == job_name:
-                    return fields[0].strip()
-        return None
+    def _get_yarn_applications_from_rm(self, resource_manager):
+        result = None
+        logging.debug('Querying list of yarn applications from %s', resource_manager)
+        try:
+            url = 'http://%s/ws/v1/cluster/apps' % resource_manager
+            result = requests.get(url, headers={'Accept': 'application/json'}).json()
+        except:
+            logging.info('Failed to query application list from %s', url)
+
+        return result
+
+    def _get_yarn_applications(self):
+        result = self._get_yarn_applications_from_rm(self._yarn_resource_manager)
+        if result is None and self._yarn_resource_manager_backup is not None:
+            result = self._get_yarn_applications_from_rm(self._yarn_resource_manager_backup)
+        return result
+
+    def _get_yarn_start_time(self, app_info):
+        try:
+            return int(app_info['startedTime'])
+        except:
+            return 0
+
+    def _find_yarn_app_info(self, all_yarn_applications, job_name):
+        result = None
+        if 'apps' in all_yarn_applications and 'app' in all_yarn_applications['apps']:
+            for app in all_yarn_applications['apps']['app']:
+                if app['name'] == job_name:
+                    if result is None or self._get_yarn_start_time(app) > self._get_yarn_start_time(result):
+                        result = app
+        return result
diff --git a/api/src/main/resources/test_application_creator.py b/api/src/main/resources/test_application_creator.py
@@ -23,7 +23,7 @@
 
 import unittest
 from datetime import datetime
-from mock import patch, mock_open
+from mock import patch, mock_open, Mock
 from application_creator import ApplicationCreator
 from exceptiondef import FailedValidation, FailedCreation
 
@@ -357,13 +357,147 @@ def test_destroy_application(self, rmdir_mock, isdir_mock, put_mock, exec_ssh_mo
         put_mock.assert_any_call('oozie/v1/job/someid2?action=kill')
 
     # pylint: disable=line-too-long
-    @patch('subprocess.check_output')
-    def test_get_runtime_details(self, subprocess_mock):
-        subprocess_mock.return_value = '''Total number of applications (application-types: [] and states: [SUBMITTED, ACCEPTED, RUNNING]):1
-                Application-Id	    Application-Name	    Application-Type	      User	     Queue	             State	       Final-State	       Progress	                       Tracking-URL
-application_1455877292606_13009	             aname-componentA-job	               SPARK	      hdfs	 root.hdfs	           RUNNING	         UNDEFINED	            10%	           http://psm-cdh-dn0:51739
-application_1455877292606_13010	             aname-componentC-job	               SPARK	      hdfs	 root.hdfs	           RUNNING	         UNDEFINED	            10%	           http://psm-cdh-dn0:51739'''
+    @patch('requests.get')
+    def test_get_runtime_details(self, get_mock):
+        rm_call = Mock()
+        rm_call.json.return_value = {
+            "apps": {
+                "app": [{
+                    "id": "application_1455877292606_13009",
+                    "user": "hdfs",
+                    "name": "aname-componentA-job",
+                    "queue": "root.users.hdfs",
+                    "state": "FINISHED",
+                    "finalStatus": "SUCCEEDED",
+                    "progress": 100.0,
+                    "trackingUI": "History",
+                    "trackingUrl": "",
+                    "diagnostics": "",
+                    "clusterId": 1479988623709,
+                    "applicationType": "MAPREDUCE",
+                    "applicationTags": "",
+                    "startedTime": 1479996060665,
+                    "finishedTime": 1479996103786,
+                    "elapsedTime": 43121,
+                    "amContainerLogs": "",
+                    "amHostHttpAddress": "",
+                    "allocatedMB": -1,
+                    "allocatedVCores": -1,
+                    "runningContainers": -1,
+                    "memorySeconds": 30600,
+                    "vcoreSeconds": 29,
+                    "preemptedResourceMB": 0,
+                    "preemptedResourceVCores": 0,
+                    "numNonAMContainerPreempted": 0,
+                    "numAMContainerPreempted": 0,
+                    "logAggregationStatus": "DISABLED"
+                }, {
+                    "id": "application_1455877292606_13010",
+                    "user": "hdfs",
+                    "name": "aname-componentC-job",
+                    "queue": "root.users.hdfs",
+                    "state": "FINISHED",
+                    "finalStatus": "SUCCEEDED",
+                    "progress": 100.0,
+                    "trackingUI": "History",
+                    "trackingUrl": "",
+                    "diagnostics": "",
+                    "clusterId": 1479988623709,
+                    "applicationType": "MAPREDUCE",
+                    "applicationTags": "",
+                    "startedTime": 1479996060665,
+                    "finishedTime": 1479996103786,
+                    "elapsedTime": 43121,
+                    "amContainerLogs": "",
+                    "amHostHttpAddress": "",
+                    "allocatedMB": -1,
+                    "allocatedVCores": -1,
+                    "runningContainers": -1,
+                    "memorySeconds": 30600,
+                    "vcoreSeconds": 29,
+                    "preemptedResourceMB": 0,
+                    "preemptedResourceVCores": 0,
+                    "numNonAMContainerPreempted": 0,
+                    "numAMContainerPreempted": 0,
+                    "logAggregationStatus": "DISABLED"
+                }, {
+                    "id": "application_1455877292606_13011",
+                    "user": "hdfs",
+                    "name": "aname-componentC-job",
+                    "queue": "root.users.hdfs",
+                    "state": "RUNNING",
+                    "finalStatus": "SUCCEEDED",
+                    "progress": 100.0,
+                    "trackingUI": "History",
+                    "trackingUrl": "",
+                    "diagnostics": "",
+                    "clusterId": 1479988623709,
+                    "applicationType": "MAPREDUCE",
+                    "applicationTags": "",
+                    "startedTime": 1479996060667,
+                    "finishedTime": 1479996103786,
+                    "elapsedTime": 43121,
+                    "amContainerLogs": "",
+                    "amHostHttpAddress": "",
+                    "allocatedMB": -1,
+                    "allocatedVCores": -1,
+                    "runningContainers": -1,
+                    "memorySeconds": 30600,
+                    "vcoreSeconds": 29,
+                    "preemptedResourceMB": 0,
+                    "preemptedResourceVCores": 0,
+                    "numNonAMContainerPreempted": 0,
+                    "numAMContainerPreempted": 0,
+                    "logAggregationStatus": "DISABLED"
+                }, {
+                    "id": "application_1455877292606_13012",
+                    "user": "hdfs",
+                    "name": "aname-componentC-job",
+                    "queue": "root.users.hdfs",
+                    "state": "NOT STARTED",
+                    "finalStatus": "SUCCEEDED",
+                    "progress": 100.0,
+                    "trackingUI": "History",
+                    "trackingUrl": "",
+                    "diagnostics": "",
+                    "clusterId": 1479988623709,
+                    "applicationType": "MAPREDUCE",
+                    "applicationTags": "",
+                    "startedTime": None,
+                    "finishedTime": 1479996103786,
+                    "elapsedTime": 43121,
+                    "amContainerLogs": "",
+                    "amHostHttpAddress": "",
+                    "allocatedMB": -1,
+                    "allocatedVCores": -1,
+                    "runningContainers": -1,
+                    "memorySeconds": 30600,
+                    "vcoreSeconds": 29,
+                    "preemptedResourceMB": 0,
+                    "preemptedResourceVCores": 0,
+                    "numNonAMContainerPreempted": 0,
+                    "numAMContainerPreempted": 0,
+                    "logAggregationStatus": "DISABLED"
+                }]
+            }
+        }
+        get_mock.return_value = rm_call
+
         creator = ApplicationCreator(self.config, self.environment, self.service)
         result = creator.get_application_runtime_details('name', self.create_data)
-        self.assertEqual(result, {'yarn_ids': [{'component': 'componentA', 'type': 'oozie', 'yarn-id': 'application_1455877292606_13009'},
-                                               {'component': 'componentC', 'type': 'sparkStreaming', 'yarn-id': 'application_1455877292606_13010'}]})
+        self.assertEqual(result, {"yarn_applications": {
+            "oozie-componentA": {
+                "type": "oozie",
+                "yarn-id": "application_1455877292606_13009",
+                "component": "componentA",
+                "yarn-start-time": 1479996060665,
+                "yarn-state": "FINISHED"
+            },
+            "sparkStreaming-componentC": {
+                "type": "sparkStreaming",
+                "yarn-id": "application_1455877292606_13011",
+                "component": "componentC",
+                "yarn-start-time": 1479996060667,
+                "yarn-state": "RUNNING"
+            }
+        }})
