Fix topology awareness configuration

Closes-Bug: #1435309
Change-Id: Iebd03c910c812a5f014625845c00f37406c367ef

sahara/plugins/mapr/base/base_cluster_configurer.py

@@ -32,6 +32,7 @@ import sahara.plugins.mapr.services.yarn.yarn as yarn
 import sahara.plugins.mapr.util.general as util
 from sahara.topology import topology_helper as th
 import sahara.utils.configs as sahara_configs
+import sahara.utils.files as files
 
 
 LOG = logging.getLogger(__name__)
@@ -127,11 +128,13 @@ class BaseConfigurer(ac.AbstractConfigurer):
         is_node_aware = context.is_node_aware
         if is_node_aware:
             topo = th.generate_topology_map(context.cluster, is_node_aware)
-            topo = '\n'.join(['%s %s' % i for i in six.iteritems(topo)])
+            topo = '\n'.join(['%s %s' % i for i in six.iteritems(topo)]) + '\n'
             data_path = '%s/topology.data' % context.mapr_home
+            script = files.get_file_text(_TOPO_SCRIPT)
+            script_path = '%s/topology.sh' % context.mapr_home
             util.execute_on_instances(instances, write_file, data_path, topo)
             util.execute_on_instances(
-                instances, util.run_script, _TOPO_SCRIPT, 'root', data_path)
+                instances, util.write_file, script_path, script, '+x', 'root')
         else:
             LOG.debug('Data locality is disabled.')
         LOG.info(_LI('Cluster topology successfully configured'))

sahara/plugins/mapr/services/mapreduce/mapreduce.py

@@ -90,19 +90,34 @@ class MapReduce(s.Service):
     def _get_core_site_props(self, context):
         result = {}
         if context.is_node_aware:
-            for conf in topo.vm_awareness_core_config():
-                result[conf['name']] = conf['value']
+            result.update(self._get_core_site_node_aware_props())
         for conf in swift_helper.get_swift_configs():
             result[conf['name']] = conf['value']
         for conf in self._get_impersonation_props():
             result[conf['name']] = conf['value']
         return result
 
+    def _get_core_site_node_aware_props(self):
+        result = topo.vm_awareness_core_config()
+        result = {c['name']: c['value'] for c in result}
+        result.update({
+            'net.topology.script.file.name': '/opt/mapr/topology.sh',
+            'net.topology.script.number.args': '75',
+        })
+        return result
+
     def _get_mapred_site_props(self, context):
         result = {}
         if context.is_node_aware:
-            for conf in topo.vm_awareness_mapred_config():
-                result[conf['name']] = conf['value']
+            result.update(self._get_mapred_site_node_aware_props())
+        return result
+
+    def _get_mapred_site_node_aware_props(self):
+        result = topo.vm_awareness_mapred_config()
+        result = {c['name']: c['value'] for c in result}
+        # This config causes failure
+        result.pop('mapred.task.cache.levels')
+        result['mapreduce.jobtracker.taskcache.levels'] = '3'
         return result
 
     def _get_impersonation_props(self):

sahara/plugins/mapr/services/maprfs/maprfs.py

@@ -130,7 +130,4 @@ class MapRFS(s.Service):
 
     def _get_cldb_conf_props(self, context):
         zookeepers = context.get_zookeeper_nodes_ip_with_port()
-        result = {'cldb.zookeeper.servers': zookeepers}
-        if context.is_node_aware:
-            result['net.topology.script.file.name'] = '/opt/mapr/topology.sh'
-        return result
+        return {'cldb.zookeeper.servers': zookeepers}

sahara/plugins/mapr/services/yarn/yarn.py

@@ -20,6 +20,7 @@ import sahara.plugins.mapr.domain.node_process as np
 import sahara.plugins.mapr.domain.service as s
 import sahara.plugins.mapr.util.validation_utils as vu
 from sahara.swift import swift_helper
+from sahara.topology import topology_helper as topo
 
 
 RESOURCE_MANAGER = np.NodeProcess(
@@ -81,10 +82,21 @@ class YARN(s.Service):
             'hadoop.proxyuser.mapr.groups': '*',
             'hadoop.proxyuser.mapr.hosts': '*',
         }
+        if context.is_node_aware:
+            result.update(self._get_core_site_node_aware_props())
         for conf in swift_helper.get_swift_configs():
             result[conf['name']] = conf['value']
         return result
 
+    def _get_core_site_node_aware_props(self):
+        result = topo.vm_awareness_core_config()
+        result = {c['name']: c['value'] for c in result}
+        result.update({
+            'net.topology.script.file.name': '/opt/mapr/topology.sh',
+            'net.topology.script.number.args': '75',
+        })
+        return result
+
     def _get_yarn_site_props(self, context):
         return {
             'hadoop.proxyuser.mapr.groups': '*',

sahara/plugins/mapr/versions/v3_1_1/context.py

@@ -26,7 +26,7 @@ class Context(bc.BaseClusterContext):
         self._hadoop_conf = None
         self._resource_manager_uri = 'maprfs:///'
         self._cluster_mode = None
-        self._node_aware = False
+        self._node_aware = True
         self._mapr_version = '3.1.1'
         self._ubuntu_ecosystem_repo = (
             'http://package.mapr.com/releases/ecosystem/ubuntu binary/')

sahara/plugins/mapr/versions/v4_0_1_mrv1/context.py

@@ -25,7 +25,7 @@ class Context(bc.BaseClusterContext):
         self._hadoop_conf = None
         self._resource_manager_uri = 'maprfs:///'
         self._cluster_mode = mr.MapReduce.cluster_mode
-        self._node_aware = False
+        self._node_aware = True
         self._mapr_version = '4.0.1'
         self._ubuntu_ecosystem_repo = (
             'http://package.mapr.com/releases/ecosystem-4.x/ubuntu binary/')

sahara/plugins/mapr/versions/v4_0_1_mrv2/context.py

@@ -25,7 +25,7 @@ class Context(bc.BaseClusterContext):
         self._hadoop_conf = None
         self._resource_manager_uri = None
         self._cluster_mode = yarn.YARNv241.cluster_mode
-        self._node_aware = False
+        self._node_aware = True
         self._mapr_version = '4.0.1'
         self._ubuntu_ecosystem_repo = (
             'http://package.mapr.com/releases/ecosystem-4.x/ubuntu binary/')