Merge branch 'develop' into PNDA-4012

dharaneeshvrd · web-flow · commit 4a0f929c2c4d · 2018-03-21T10:42:29.000+05:30
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,10 @@
 All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
+### Changed
+- PNDA-4405: Require user and package fields in create application API
+- PNDA-4389: Reject packages containing upstart.conf files
+
 ### Fixed
 - PNDA-4218: Fix application deletion when app name is an HDFS username
 
diff --git a/api/src/main/resources/app.py b/api/src/main/resources/app.py
@@ -256,13 +256,20 @@ class ApplicationHandler(BaseHandler):
     def put(self, aname):
         try:
             request_body = json.loads(self.request.body)
-            pname = request_body['package']
-        except:
+        except ValueError:
             self.send_client_error("Invalid request body")
             return
 
+        if 'package' not in request_body:
+            self.send_client_error("Invalid request body. Missing field 'package'")
+            return
+
+        if 'user' not in request_body:
+            self.send_client_error("Invalid request body. Missing field 'user'")
+            return
+
         def do_call():
-            dm.create_application(pname, aname, request_body)
+            dm.create_application(request_body['package'], aname, request_body)
             self.send_accepted()
 
         DISPATCHER.run_as_asynch(task=do_call, on_error=self.handle_error)
diff --git a/api/src/main/resources/application_summary.py b/api/src/main/resources/application_summary.py
diff --git a/api/src/main/resources/deployer_utils.py b/api/src/main/resources/deployer_utils.py
@@ -213,7 +213,7 @@ def fill_hadoop_env_cdh(env):
                 if not nameservice and role.type == "NAMENODE":
                     env['name_node'] = 'hdfs://%s:8020' % api.get_host(role.hostRef.hostId).hostname
                 if role.type == "HTTPFS":
-                    env['webhdfs_host'] = '%s' % api.get_host(role.hostRef.hostId).ipAddress
+                    env['webhdfs_host'] = '%s' % api.get_host(role.hostRef.hostId).hostname
                     env['webhdfs_port'] = '14000'
         elif service.type == "YARN":
             for role in service.get_all_roles():
diff --git a/api/src/main/resources/plugins/oozie.py b/api/src/main/resources/plugins/oozie.py
@@ -22,11 +22,14 @@
 
 # pylint: disable=C0103
 
+import os
 import json
 import logging
 import datetime
 import xml.etree.ElementTree as ElementTree
 import commands
+import shutil
+import traceback
 import requests
 
 import deployer_utils
@@ -111,41 +114,28 @@ def create_component(self, staged_component_path, application_name, user_name, c
         properties[def_path] = '%s/%s' % (self._environment['name_node'], remote_path)
 
         # deploy everything to various hadoop services
-        undeploy = self._deploy_to_hadoop(properties, staged_component_path, remote_path, properties['application_user'])
+        undeploy = self._deploy_to_hadoop(component, properties, staged_component_path, remote_path, properties['application_user'])
 
         # return something that can be used to undeploy later
         return {'job_handle': undeploy['id'],
                 'component_hdfs_root': properties['component_hdfs_root'],
                 'application_user': properties['application_user']}
 
-    def _deploy_to_hadoop(self, properties, staged_component_path, remote_path, application_user, exclude=None):
-        if exclude is None:
-            exclude = []
-        exclude.extend(['hdfs.json',
-                        'hbase.json',
-                        'properties.json',
-                        'application.properties'])
-
-        # stage the component files to hdfs
-        self._hdfs_client.recursive_copy(staged_component_path, remote_path, exclude=exclude, permission=755)
-
-        # stage the instantiated job properties back to HDFS - no functional purpose,
-        # just helps developers understand what has happened
-        effective_job_properties = deployer_utils.dict_to_props(properties)
-        self._hdfs_client.create_file(effective_job_properties, '%s/application.properties' % remote_path)
-
+    def _setup_queue_config(self, component, staged_component_path, properties):
         # Add queue config into the default config if none is defined.
         if 'mapreduce.job.queuename' in properties:
             defaults = {'mapreduce.job.queuename':properties['mapreduce.job.queuename']}
             try:
-                data = self._hdfs_client.read_file('%s/config-default.xml' % remote_path)
+                with open('%s/config-default.xml' % staged_component_path, 'r') as config_default_file:
+                    data = config_default_file.read()
             except:
                 logging.debug('No config-default.xml is detected.')
                 data = None
 
             if data is None:
                 logging.debug('Creating config-default.xml to inject mapreduce.job.queuename property.')
-                self._hdfs_client.create_file(deployer_utils.dict_to_xml(defaults), '%s/config-default.xml' % remote_path)
+                with open('%s/config-default.xml' % staged_component_path, 'w') as config_default_file:
+                    config_default_file.write(deployer_utils.dict_to_xml(defaults))
             else:
                 prop = None
                 root = None
@@ -167,10 +157,76 @@ def _deploy_to_hadoop(self, properties, staged_component_path, remote_path, appl
                         logging.debug('adding mapred.queue.names in config-default.xml')
                         prop = ElementTree.SubElement(root, 'property')
                         ElementTree.SubElement(prop, 'name').text = 'mapreduce.job.queuename'
-                        ElementTree.SubElement(prop, 'value').text = 'dev'
+                        ElementTree.SubElement(prop, 'value').text = properties['mapreduce.job.queuename']
                         data = ElementTree.tostring(root)
-                        self._hdfs_client.remove('%s/config-default.xml' % remote_path)
-                        self._hdfs_client.create_file(data, '%s/config-default.xml' % remote_path)
+                        with open('%s/config-default.xml' % staged_component_path, 'w') as config_default_file:
+                            config_default_file.write(data)
+
+            file_list = [file_name for file_name in component['component_detail'] if os.path.isfile('%s/%s' % (staged_component_path, file_name))]
+            # find workflow.xml files
+            for afile in file_list:
+                workflow_modified = False
+                file_path = '%s/%s' % (staged_component_path, afile)
+                with open(file_path, 'r') as component_file:
+                    workflow_xml = component_file.read()
+                    if 'uri:oozie:workflow' not in workflow_xml:
+                        continue
+                logging.debug("Found workflow file %s", file_path)
+                # copy config-default.xml into this directory
+                if os.path.dirname(file_path) != staged_component_path:
+                    shutil.copyfile('%s/config-default.xml' % staged_component_path, '%s/config-default.xml' % os.path.dirname(file_path))
+
+                # set the spark opts --queue so spark jobs are put in the right queue
+                spark_action_index = 0
+                while spark_action_index >= 0:
+                    spark_action_index = workflow_xml.find('<spark ', spark_action_index+1)
+                    spark_end_index = workflow_xml.find('</spark>', spark_action_index)
+                    jar_end_index = workflow_xml.find('</jar>', spark_action_index, spark_end_index)
+                    opts_index = workflow_xml.find('<spark-opts>', spark_action_index, spark_end_index)
+                    opts_end_index = workflow_xml.find('</spark-opts>', opts_index, spark_end_index)
+                    queue_opt_index = workflow_xml.find('--queue ', opts_index, opts_end_index)
+                    if jar_end_index >= 0:
+                        if opts_index < 0:
+                            # we need to add a spark-opts element
+                            split_index = jar_end_index+len('</jar>')
+                            workflow_xml = '%s%s%s' % (workflow_xml[:split_index],
+                                                       '<spark-opts>--queue ${wf:conf("mapreduce.job.queuename")}</spark-opts>',
+                                                       workflow_xml[split_index:])
+                            workflow_modified = True
+                        elif queue_opt_index < 0:
+                            # we need to add a queue opt to the existing spark-opts element
+                            split_index = opts_end_index
+                            workflow_xml = '%s%s%s' % (workflow_xml[:split_index], ' --queue ${wf:conf("mapreduce.job.queuename")}', workflow_xml[split_index:])
+                            workflow_modified = True
+
+                # write out modified workflow if changes were made
+                if workflow_modified:
+                    logging.debug("Writing out modified workflow xml to %s", file_path)
+                    with open(file_path, "w") as workflow_file:
+                        workflow_file.write(workflow_xml)
+
+    def _deploy_to_hadoop(self, component, properties, staged_component_path, remote_path, application_user, exclude=None):
+        if exclude is None:
+            exclude = []
+        exclude.extend(['hdfs.json',
+                        'hbase.json',
+                        'properties.json',
+                        'application.properties'])
+
+        # setup queue config
+        try:
+            self._setup_queue_config(component, staged_component_path, properties)
+        except Exception as ex:
+            logging.error(traceback.format_exc())
+            raise FailedCreation('Failed to set up yarn queue config: %s' % str(ex))
+
+        # stage the component files to hdfs
+        self._hdfs_client.recursive_copy(staged_component_path, remote_path, exclude=exclude, permission=755)
+
+        # stage the instantiated job properties back to HDFS - no functional purpose,
+        # just helps developers understand what has happened
+        effective_job_properties = deployer_utils.dict_to_props(properties)
+        self._hdfs_client.create_file(effective_job_properties, '%s/application.properties' % remote_path)
 
         # submit to oozie
         result = self._submit_oozie(properties)
diff --git a/api/src/main/resources/plugins/sparkStreaming.py b/api/src/main/resources/plugins/sparkStreaming.py
@@ -40,6 +40,10 @@ def validate_component(self, component):
             errors.append('missing file application.properties')
         if 'log4j.properties' not in file_list:
             errors.append('missing file log4j.properties')
+        if 'upstart.conf' in file_list:
+            errors.append('Support for user supplied upstart.conf files has been deprecated, ' +
+                          'the deployment manager will supply one automatically. ' +
+                          'Please see PNDA example-applications for usage.')
         return errors
 
     def get_component_type(self):
@@ -85,9 +89,10 @@ def create_component(self, staged_component_path, application_name, user_name, c
             properties['component_py_files'] = ''
 
         if 'upstart.conf' in component['component_detail']:
-            # old style applications for backward compatibility
-            service_script = 'upstart.conf'
-            service_script_install_path = '/etc/init/%s.conf' % service_name
+            # old style applications - reject these
+            raise Exception('Support for user supplied upstart.conf files has been deprecated, ' +
+                             'the deployment manager will supply one automatically. ' +
+                             'Please see PNDA example-applications for usage.')
         else:
             # new style applications that don't need to provide upstart.conf or yarn-kill.py
             if 'component_main_jar' in properties and 'component_main_class' not in properties:
diff --git a/api/src/main/resources/requirements.txt b/api/src/main/resources/requirements.txt
@@ -1,7 +1,7 @@
 argparse==1.2.1
 backports-abc==0.5
 backports.ssl-match-hostname==3.5.0.1
-certifi==2016.9.26
+certifi==2017.4.17
 cffi==1.9.1
 cm-api==14.0.0
 cryptography==1.7.1
@@ -26,4 +26,4 @@ thrift==0.9.3
 thriftpy==0.3.9
 tornado==4.4.2
 tornado-cors==0.6.0
-wsgiref==0.1.2
+wsgiref==0.1.2
diff --git a/api/src/main/resources/test_application_summary.py b/api/src/main/resources/test_application_summary.py
