Merge "Formatting and mounting methods changed for ironic"

sahara/conductor/manager.py

@@ -55,7 +55,8 @@ NODE_GROUP_TEMPLATE_DEFAULTS.update({"is_public": False,
                                      "is_protected": False})
 
 INSTANCE_DEFAULTS = {
-    "volumes": []
+    "volumes": [],
+    "storage_devices_number": 0
 }
 
 DATA_SOURCE_DEFAULTS = {

sahara/conductor/objects.py

@@ -131,18 +131,6 @@ class NodeGroup(object):
         return configs.merge_configs(self.cluster.cluster_configs,
                                      self.node_configs)
 
-    def storage_paths(self):
-        mp = []
-        for idx in range(1, self.volumes_per_node + 1):
-            mp.append(self.volume_mount_prefix + str(idx))
-
-        # Here we assume that NG's instances use ephemeral
-        # drives for storage if volumes_per_node == 0
-        if not mp:
-            mp = ['/mnt']
-
-        return mp
-
     def get_image_id(self):
         return self.image_id or self.cluster.default_image_id
 
@@ -158,6 +146,7 @@ class Instance(object):
     internal_ip
     management_ip
     volumes
+    storage_devices_number
     """
 
     def hostname(self):
@@ -169,6 +158,16 @@ class Instance(object):
     def remote(self):
         return remote.get_remote(self)
 
+    def storage_paths(self):
+        mp = []
+        for idx in range(1, self.storage_devices_number + 1):
+            mp.append(self.node_group.volume_mount_prefix + str(idx))
+
+        if not mp:
+            mp = ['/mnt']
+
+        return mp
+
 
 class ClusterTemplate(object):
     """An object representing Cluster Template.

sahara/db/migration/alembic_migrations/versions/028_storage_devices_number.py

@@ -0,0 +1,35 @@
+# Copyright 2015 OpenStack Foundation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""add_storage_devices_number
+
+Revision ID: 028
+Revises: 027
+Create Date: 2015-07-20 16:56:23.562710
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '028'
+down_revision = '027'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    op.add_column('instances',
+                  sa.Column('storage_devices_number', sa.Integer(),
+                            nullable=True))

sahara/db/sqlalchemy/models.py

@@ -157,6 +157,7 @@ class Instance(mb.SaharaBase):
     internal_ip = sa.Column(sa.String(45))
     management_ip = sa.Column(sa.String(45))
     volumes = sa.Column(st.JsonListType())
+    storage_devices_number = sa.Column(sa.Integer)
 
 
 # Template objects: ClusterTemplate, NodeGroupTemplate, TemplatesRelation

sahara/plugins/ambari/configs.py

@@ -167,10 +167,10 @@ def _make_paths(dirs, suffix):
     return ",".join([d + suffix for d in dirs])
 
 
-def get_ng_params(node_group):
+def get_instance_params(inst):
-    configs = _create_ambari_configs(node_group.node_configs,
+    configs = _create_ambari_configs(inst.node_group.node_configs,
-                                     node_group.cluster.hadoop_version)
+                                     inst.node_group.cluster.hadoop_version)
-    storage_paths = node_group.storage_paths()
+    storage_paths = inst.storage_paths()
     configs.setdefault("hdfs-site", {})
     configs["hdfs-site"]["dfs.datanode.data.dir"] = _make_paths(
         storage_paths, "/hdfs/data")
@@ -188,7 +188,6 @@ def get_ng_params(node_group):
     configs["yarn-site"][
         "yarn.timeline-service.leveldb-timeline-store.path"] = _make_paths(
             storage_paths, "/yarn/timeline")
-
     return _serialize_ambari_configs(configs)
 
 

sahara/plugins/ambari/deploy.py

@@ -183,13 +183,14 @@ def create_blueprint(cluster):
     cluster = conductor.cluster_get(context.ctx(), cluster.id)
     host_groups = []
     for ng in cluster.node_groups:
-        hg = {
-            "name": ng.name,
-            "configurations": configs.get_ng_params(ng),
-            "components": []
-        }
         procs = p_common.get_ambari_proc_list(ng)
         procs.extend(p_common.get_clients(cluster))
+        for instance in ng.instances:
+            hg = {
+                "name": instance.instance_name,
+                "configurations": configs.get_instance_params(instance),
+                "components": []
+            }
             for proc in procs:
                 hg["components"].append({"name": proc})
             host_groups.append(hg)
@@ -214,9 +215,10 @@ def start_cluster(cluster):
         "host_groups": []
     }
     for ng in cluster.node_groups:
+        for instance in ng.instances:
             cl_tmpl["host_groups"].append({
-            "name": ng.name,
+                "name": instance.instance_name,
-            "hosts": map(lambda x: {"fqdn": x.fqdn()}, ng.instances)
+                "hosts": [{"fqdn": instance.fqdn()}]
             })
     ambari = plugin_utils.get_instance(cluster, p_common.AMBARI_SERVER)
     password = cluster.extra["ambari_password"]

sahara/plugins/cdh/cloudera_utils.py

@@ -284,7 +284,7 @@ class ClouderaUtils(object):
         role = service.create_role(self.pu.get_role_name(instance, process),
                                    role_type, instance.fqdn())
         role.update_config(self._get_configs(process, cluster,
-                                             node_group=instance.node_group))
+                                             instance=instance))
 
     def get_cloudera_manager_info(self, cluster):
         mng = self.pu.get_manager(cluster)
@@ -297,6 +297,6 @@ class ClouderaUtils(object):
         }
         return info
 
-    def _get_configs(self, service, cluster=None, node_group=None):
+    def _get_configs(self, service, cluster=None, instance=None):
         # Defined in derived class.
         return

sahara/plugins/cdh/v5/cloudera_utils.py

@@ -145,7 +145,7 @@ class ClouderaUtilsV5(cu.ClouderaUtils):
             hbase.update_config(self._get_configs(HBASE_SERVICE_TYPE,
                                                   cluster=cluster))
 
-    def _get_configs(self, service, cluster=None, node_group=None):
+    def _get_configs(self, service, cluster=None, instance=None):
         def get_hadoop_dirs(mount_points, suffix):
             return ','.join([x + suffix for x in mount_points])
 
@@ -214,10 +214,10 @@ class ClouderaUtilsV5(cu.ClouderaUtils):
             all_confs = s_cfg.merge_configs(all_confs, hive_confs)
             all_confs = s_cfg.merge_configs(all_confs, cluster.cluster_configs)
 
-        if node_group:
+        if instance:
-            paths = node_group.storage_paths()
+            paths = instance.storage_paths()
 
-            ng_default_confs = {
+            instance_default_confs = {
                 'NAMENODE': {
                     'dfs_name_dir_list': get_hadoop_dirs(paths, '/fs/nn')
                 },
@@ -240,8 +240,8 @@ class ClouderaUtilsV5(cu.ClouderaUtils):
             }
 
             ng_user_confs = self.pu.convert_process_configs(
-                node_group.node_configs)
+                instance.node_group.node_configs)
             all_confs = s_cfg.merge_configs(all_confs, ng_user_confs)
-            all_confs = s_cfg.merge_configs(all_confs, ng_default_confs)
+            all_confs = s_cfg.merge_configs(all_confs, instance_default_confs)
 
         return all_confs.get(service, {})

sahara/plugins/cdh/v5_3_0/cloudera_utils.py

@@ -205,7 +205,7 @@ class ClouderaUtilsV530(cu.ClouderaUtils):
             impala.update_config(self._get_configs(IMPALA_SERVICE_TYPE,
                                                    cluster=cluster))
 
-    def _get_configs(self, service, cluster=None, node_group=None):
+    def _get_configs(self, service, cluster=None, instance=None):
         def get_hadoop_dirs(mount_points, suffix):
             return ','.join([x + suffix for x in mount_points])
 
@@ -344,10 +344,10 @@ class ClouderaUtilsV530(cu.ClouderaUtils):
             all_confs = s_cfg.merge_configs(all_confs, sentry_confs)
             all_confs = s_cfg.merge_configs(all_confs, cluster.cluster_configs)
 
-        if node_group:
+        if instance:
-            paths = node_group.storage_paths()
+            paths = instance.storage_paths()
 
-            ng_default_confs = {
+            instance_default_confs = {
                 'NAMENODE': {
                     'dfs_name_dir_list': get_hadoop_dirs(paths, '/fs/nn')
                 },
@@ -382,8 +382,8 @@ class ClouderaUtilsV530(cu.ClouderaUtils):
             }
 
             ng_user_confs = self.pu.convert_process_configs(
-                node_group.node_configs)
+                instance.node_group.node_configs)
             all_confs = s_cfg.merge_configs(all_confs, ng_user_confs)
-            all_confs = s_cfg.merge_configs(all_confs, ng_default_confs)
+            all_confs = s_cfg.merge_configs(all_confs, instance_default_confs)
 
         return all_confs.get(service, {})

sahara/plugins/cdh/v5_4_0/cloudera_utils.py

@@ -222,7 +222,7 @@ class ClouderaUtilsV540(cu.ClouderaUtils):
             kms.update_config(self._get_configs(KMS_SERVICE_TYPE,
                                                 cluster=cluster))
 
-    def _get_configs(self, service, cluster=None, node_group=None):
+    def _get_configs(self, service, cluster=None, instance=None):
         def get_hadoop_dirs(mount_points, suffix):
             return ','.join([x + suffix for x in mount_points])
 
@@ -363,10 +363,10 @@ class ClouderaUtilsV540(cu.ClouderaUtils):
             all_confs = s_cfg.merge_configs(all_confs, sentry_confs)
             all_confs = s_cfg.merge_configs(all_confs, cluster.cluster_configs)
 
-        if node_group:
+        if instance:
-            paths = node_group.storage_paths()
+            paths = instance.storage_paths()
 
-            ng_default_confs = {
+            instance_default_confs = {
                 'NAMENODE': {
                     'dfs_name_dir_list': get_hadoop_dirs(paths, '/fs/nn')
                 },
@@ -401,9 +401,9 @@ class ClouderaUtilsV540(cu.ClouderaUtils):
             }
 
             ng_user_confs = self.pu.convert_process_configs(
-                node_group.node_configs)
+                instance.node_group.node_configs)
             all_confs = s_cfg.merge_configs(all_confs, ng_user_confs)
-            all_confs = s_cfg.merge_configs(all_confs, ng_default_confs)
+            all_confs = s_cfg.merge_configs(all_confs, instance_default_confs)
 
         return all_confs.get(service, {})
 

sahara/plugins/hdp/clusterspec.py

@@ -210,7 +210,6 @@ class ClusterSpec(object):
                 node_group.count = ng.count
                 node_group.id = ng.id
                 node_group.components = ng.node_processes[:]
-                node_group.ng_storage_paths = ng.storage_paths()
                 for instance in ng.instances:
                     node_group.instances.add(Instance(instance))
                 self.node_groups[node_group.name] = node_group
@@ -270,14 +269,10 @@ class NodeGroup(object):
         self.cardinality = None
         self.count = None
         self.instances = set()
-        self.ng_storage_paths = []
 
     def add_component(self, component):
         self.components.append(component)
 
-    def storage_paths(self):
-        return self.ng_storage_paths
-
 
 class User(object):
     def __init__(self, name, password, groups):

sahara/plugins/hdp/versions/version_2_0_6/services.py

@@ -102,16 +102,12 @@ class Service(object):
             config[prop_name] = value
 
     def _get_common_paths(self, node_groups):
-        if len(node_groups) == 1:
+        sets = []
-            paths = node_groups[0].storage_paths()
+        for node_group in node_groups:
-        else:
+            for instance in node_group.instances:
-            sets = [set(ng.storage_paths()) for ng in node_groups]
+                sets.append(set(instance.sahara_instance.storage_paths()))
-            paths = list(set.intersection(*sets))
 
-        if len(paths) > 1 and '/mnt' in paths:
+        return list(set.intersection(*sets)) if sets else []
-            paths.remove('/mnt')
-
-        return paths
 
     def _generate_storage_path(self, storage_paths, path):
         return ",".join([p + path for p in storage_paths])
@@ -201,7 +197,7 @@ class HdfsService(Service):
         hdfs_site_config = cluster_spec.configurations['hdfs-site']
         hdfs_site_config['dfs.namenode.name.dir'] = (
             self._generate_storage_path(
-                nn_ng.storage_paths(), '/hadoop/hdfs/namenode'))
+                self._get_common_paths([nn_ng]), '/hadoop/hdfs/namenode'))
         if common_paths:
             hdfs_site_config['dfs.datanode.data.dir'] = (
                 self._generate_storage_path(

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

@@ -107,7 +107,7 @@ class MapRFS(s.Service):
     def _generate_disk_list_file(self, instance, path_to_disk_setup_script):
         LOG.debug('Creating disk list file')
         g.run_script(instance, path_to_disk_setup_script, 'root',
-                     *instance.node_group.storage_paths())
+                     *instance.storage_paths())
 
     def _execute_disksetup(self, instance):
         with instance.remote() as rmt:

sahara/plugins/spark/plugin.py

@@ -163,7 +163,6 @@ class SparkProvider(p.ProvisioningPluginBase):
                                                  cluster)
 
     def _extract_configs_to_extra(self, cluster):
-        nn = utils.get_instance(cluster, "namenode")
         sp_master = utils.get_instance(cluster, "master")
         sp_slaves = utils.get_instances(cluster, "slave")
 
@@ -186,22 +185,10 @@ class SparkProvider(p.ProvisioningPluginBase):
         config_defaults = c_helper.generate_spark_executor_classpath(cluster)
 
         extra['job_cleanup'] = c_helper.generate_job_cleanup_config(cluster)
-        for ng in cluster.node_groups:
+
-            extra[ng.id] = {
+        extra['sp_master'] = config_master
-                'xml': c_helper.generate_xml_configs(
+        extra['sp_slaves'] = config_slaves
-                    ng.configuration(),
+        extra['sp_defaults'] = config_defaults
-                    ng.storage_paths(),
-                    nn.hostname(), None
-                ),
-                'setup_script': c_helper.generate_hadoop_setup_script(
-                    ng.storage_paths(),
-                    c_helper.extract_hadoop_environment_confs(
-                        ng.configuration())
-                ),
-                'sp_master': config_master,
-                'sp_slaves': config_slaves,
-                'sp_defaults': config_defaults
-            }
 
         if c_helper.is_data_locality_enabled(cluster):
             topology_data = th.generate_topology_map(
@@ -211,6 +198,19 @@ class SparkProvider(p.ProvisioningPluginBase):
 
         return extra
 
+    def _add_instance_ng_related_to_extra(self, cluster, instance, extra):
+        extra = extra.copy()
+        ng = instance.node_group
+        nn = utils.get_instance(cluster, "namenode")
+
+        extra['xml'] = c_helper.generate_xml_configs(
+            ng.configuration(), instance.storage_paths(), nn.hostname(), None)
+        extra['setup_script'] = c_helper.generate_hadoop_setup_script(
+            instance.storage_paths(),
+            c_helper.extract_hadoop_environment_confs(ng.configuration()))
+
+        return extra
+
     def _start_datanode_processes(self, dn_instances):
         if len(dn_instances) == 0:
             return
@@ -243,6 +243,8 @@ class SparkProvider(p.ProvisioningPluginBase):
             cluster.id, _("Push configs to nodes"), len(all_instances))
         with context.ThreadGroup() as tg:
             for instance in all_instances:
+                extra = self._add_instance_ng_related_to_extra(
+                    cluster, instance, extra)
                 if instance in new_instances:
                     tg.spawn('spark-configure-%s' % instance.instance_name,
                              self._push_configs_to_new_node, cluster,
@@ -254,25 +256,23 @@ class SparkProvider(p.ProvisioningPluginBase):
 
     @cpo.event_wrapper(mark_successful_on_exit=True)
     def _push_configs_to_new_node(self, cluster, extra, instance):
-        ng_extra = extra[instance.node_group.id]
-
         files_hadoop = {
             os.path.join(c_helper.HADOOP_CONF_DIR,
-                         "core-site.xml"): ng_extra['xml']['core-site'],
+                         "core-site.xml"): extra['xml']['core-site'],
             os.path.join(c_helper.HADOOP_CONF_DIR,
-                         "hdfs-site.xml"): ng_extra['xml']['hdfs-site'],
+                         "hdfs-site.xml"): extra['xml']['hdfs-site'],
         }
 
         sp_home = self._spark_home(cluster)
         files_spark = {
-            os.path.join(sp_home, 'conf/spark-env.sh'): ng_extra['sp_master'],
+            os.path.join(sp_home, 'conf/spark-env.sh'): extra['sp_master'],
-            os.path.join(sp_home, 'conf/slaves'): ng_extra['sp_slaves'],
+            os.path.join(sp_home, 'conf/slaves'): extra['sp_slaves'],
             os.path.join(sp_home,
-                         'conf/spark-defaults.conf'): ng_extra['sp_defaults']
+                         'conf/spark-defaults.conf'): extra['sp_defaults']
         }
 
         files_init = {
-            '/tmp/sahara-hadoop-init.sh': ng_extra['setup_script'],
+            '/tmp/sahara-hadoop-init.sh': extra['setup_script'],
             'id_rsa': cluster.management_private_key,
             'authorized_keys': cluster.management_public_key
         }
@@ -283,7 +283,7 @@ class SparkProvider(p.ProvisioningPluginBase):
                    'sudo chown $USER $HOME/.ssh/id_rsa; '
                    'sudo chmod 600 $HOME/.ssh/id_rsa')
 
-        storage_paths = instance.node_group.storage_paths()
+        storage_paths = instance.storage_paths()
         dn_path = ' '.join(c_helper.make_hadoop_path(storage_paths,
                                                      '/dfs/dn'))
         nn_path = ' '.join(c_helper.make_hadoop_path(storage_paths,
@@ -336,15 +336,14 @@ class SparkProvider(p.ProvisioningPluginBase):
                              'slave' in node_processes)
 
         if need_update_spark:
-            ng_extra = extra[instance.node_group.id]
             sp_home = self._spark_home(cluster)
             files = {
                 os.path.join(sp_home,
-                             'conf/spark-env.sh'): ng_extra['sp_master'],
+                             'conf/spark-env.sh'): extra['sp_master'],
-                os.path.join(sp_home, 'conf/slaves'): ng_extra['sp_slaves'],
+                os.path.join(sp_home, 'conf/slaves'): extra['sp_slaves'],
                 os.path.join(
                     sp_home,
-                    'conf/spark-defaults.conf'): ng_extra['sp_defaults']
+                    'conf/spark-defaults.conf'): extra['sp_defaults']
             }
             r = remote.get_remote(instance)
             r.write_files_to(files)

sahara/plugins/vanilla/hadoop2/config.py

@@ -73,24 +73,25 @@ def _configure_instance(pctx, instance):
 
 
 def _provisioning_configs(pctx, instance):
-    xmls, env = _generate_configs(pctx, instance.node_group)
+    xmls, env = _generate_configs(pctx, instance)
     _push_xml_configs(instance, xmls)
     _push_env_configs(instance, env)
 
 
-def _generate_configs(pctx, node_group):
+def _generate_configs(pctx, instance):
-    hadoop_xml_confs = _get_hadoop_configs(pctx, node_group)
+    hadoop_xml_confs = _get_hadoop_configs(pctx, instance)
-    user_xml_confs, user_env_confs = _get_user_configs(pctx, node_group)
+    user_xml_confs, user_env_confs = _get_user_configs(
+        pctx, instance.node_group)
     xml_confs = s_cfg.merge_configs(user_xml_confs, hadoop_xml_confs)
     env_confs = s_cfg.merge_configs(pctx['env_confs'], user_env_confs)
 
     return xml_confs, env_confs
 
 
-def _get_hadoop_configs(pctx, node_group):
+def _get_hadoop_configs(pctx, instance):
-    cluster = node_group.cluster
+    cluster = instance.node_group.cluster
     nn_hostname = vu.get_instance_hostname(vu.get_namenode(cluster))
-    dirs = _get_hadoop_dirs(node_group)
+    dirs = _get_hadoop_dirs(instance)
     confs = {
         'Hadoop': {
             'fs.defaultFS': 'hdfs://%s:9000' % nn_hostname
@@ -282,8 +283,7 @@ def _push_configs_to_instance(instance, configs):
 
 
 def _post_configuration(pctx, instance):
-    node_group = instance.node_group
+    dirs = _get_hadoop_dirs(instance)
-    dirs = _get_hadoop_dirs(node_group)
     args = {
         'hadoop_user': HADOOP_USER,
         'hadoop_group': HADOOP_GROUP,
@@ -313,9 +313,9 @@ def _post_configuration(pctx, instance):
             r.execute_command('chmod +x ' + t_script, run_as_root=True)
 
 
-def _get_hadoop_dirs(node_group):
+def _get_hadoop_dirs(instance):
     dirs = {}
-    storage_paths = node_group.storage_paths()
+    storage_paths = instance.storage_paths()
     dirs['hadoop_name_dirs'] = _make_hadoop_paths(
         storage_paths, '/hdfs/namenode')
     dirs['hadoop_data_dirs'] = _make_hadoop_paths(

sahara/service/volumes.py

@@ -13,6 +13,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import re
+import threading
+
 from oslo_config import cfg
 from oslo_log import log as logging
 
@@ -21,6 +24,7 @@ from sahara import context
 from sahara import exceptions as ex
 from sahara.i18n import _
 from sahara.i18n import _LE
+from sahara.i18n import _LW
 from sahara.utils import cluster_progress_ops as cpo
 from sahara.utils.openstack import base as b
 from sahara.utils.openstack import cinder
@@ -51,6 +55,7 @@ def _count_volumes_to_mount(instances):
 def attach_to_instances(instances):
     instances_to_attach = _count_instances_to_attach(instances)
     if instances_to_attach == 0:
+        mount_to_instances(instances)
         return
 
     cpo.add_provisioning_step(
@@ -66,6 +71,8 @@ def attach_to_instances(instances):
                         % instance.instance_name, _attach_volumes_to_node,
                         instance.node_group, instance)
 
+    mount_to_instances(instances)
+
 
 @poll_utils.poll_status(
     'await_attach_volumes', _("Await for attaching volumes to instances"),
@@ -91,13 +98,6 @@ def _attach_volumes_to_node(node_group, instance):
 
     _await_attach_volumes(instance, devices)
 
-    paths = instance.node_group.storage_paths()
-    for idx in range(0, instance.node_group.volumes_per_node):
-        LOG.debug("Mounting volume {volume} to instance"
-                  .format(volume=devices[idx]))
-        _mount_volume(instance, devices[idx], paths[idx])
-        LOG.debug("Mounted volume to instance")
-
 
 @poll_utils.poll_status(
     'volume_available_timeout', _("Await for volume become available"),
@@ -155,51 +155,90 @@ def mount_to_instances(instances):
         instances[0].cluster_id,
         _("Mount volumes to instances"), _count_volumes_to_mount(instances))
 
-    with context.ThreadGroup() as tg:
     for instance in instances:
         with context.set_current_instance_id(instance.instance_id):
-                devices = _find_instance_volume_devices(instance)
+            devices = _find_instance_devices(instance)
+            formatted_devices = []
+            lock = threading.Lock()
+            with context.ThreadGroup() as tg:
                 # Since formating can take several minutes (for large disks)
                 # and can be done in parallel, launch one thread per disk.
-                for idx in range(0, instance.node_group.volumes_per_node):
+                for device in devices:
-                    tg.spawn(
+                    tg.spawn('format-device-%s' % device, _format_device,
-                        'mount-volume-%d-to-node-%s' %
+                             instance, device, formatted_devices, lock)
-                        (idx, instance.instance_name),
+
-                        _mount_volume_to_node, instance, idx, devices[idx])
+            conductor.instance_update(
+                context.current(), instance,
+                {"storage_devices_number": len(formatted_devices)})
+            for idx, dev in enumerate(formatted_devices):
+                _mount_volume_to_node(instance, idx+1, dev)
 
 
-def _find_instance_volume_devices(instance):
+def _find_instance_devices(instance):
-    volumes = b.execute_with_retries(nova.client().volumes.get_server_volumes,
+    with instance.remote() as r:
-                                     instance.instance_id)
+        code, attached_info = r.execute_command(
-    devices = [volume.device for volume in volumes]
+            "lsblk -r | awk '$6 ~ /disk/ || /part/ {print \"/dev/\" $1}'")
-    return devices
+        attached_dev = attached_info.split()
+        code, mounted_info = r.execute_command(
+            "mount | awk '$1 ~ /^\/dev/ {print $1}'")
+        mounted_dev = mounted_info.split()
+
+        # filtering attached devices, that should not be mounted
+        for dev in attached_dev[:]:
+            idx = re.sub("\D", "", dev)
+            if idx:
+                if dev in mounted_dev:
+                    attached_dev.remove(re.sub("\d", "", dev))
+                    attached_dev.remove(dev)
+
+        for dev in attached_dev[:]:
+            if re.sub("\D", "", dev):
+                if re.sub("\d", "", dev) in attached_dev:
+                    attached_dev.remove(dev)
+
+    return attached_dev
 
 
 @cpo.event_wrapper(mark_successful_on_exit=True)
-def _mount_volume_to_node(instance, idx, device):
+def _mount_volume_to_node(instance, index, device):
     LOG.debug("Mounting volume {device} to instance".format(device=device))
-    mount_point = instance.node_group.storage_paths()[idx]
+    mount_point = instance.node_group.volume_mount_prefix + str(index)
     _mount_volume(instance, device, mount_point)
     LOG.debug("Mounted volume to instance")
 
 
+def _format_device(instance, device, formatted_devices=None, lock=None):
+    with instance.remote() as r:
+        try:
+            # Format devices with better performance options:
+            # - reduce number of blocks reserved for root to 1%
+            # - use 'dir_index' for faster directory listings
+            # - use 'extents' to work faster with large files
+            # - disable journaling
+            fs_opts = '-F -m 1 -O dir_index,extents,^has_journal'
+            r.execute_command('sudo mkfs.ext4 %s %s' % (fs_opts, device))
+            if lock:
+                with lock:
+                    formatted_devices.append(device)
+        except Exception:
+            LOG.warning(
+                _LW("Device {dev} cannot be formatted").format(dev=device))
+
+
 def _mount_volume(instance, device_path, mount_point):
     with instance.remote() as r:
         try:
             # Mount volumes with better performance options:
-            # - reduce number of blocks reserved for root to 1%
-            # - use 'dir_index' for faster directory listings
-            # - use 'extents' to work faster with large files
-            # - disable journaling
             # - enable write-back
             # - do not store access time
-            fs_opts = '-m 1 -O dir_index,extents,^has_journal'
             mount_opts = '-o data=writeback,noatime,nodiratime'
 
             r.execute_command('sudo mkdir -p %s' % mount_point)
-            r.execute_command('sudo mkfs.ext4 %s %s' % (fs_opts, device_path))
             r.execute_command('sudo mount %s %s %s' %
                               (mount_opts, device_path, mount_point))
+            r.execute_command(
+                'sudo sh -c "grep %s /etc/mtab >> /etc/fstab"' % device_path)
+
         except Exception:
             LOG.error(_LE("Error mounting volume to instance"))
             raise

sahara/tests/unit/db/migration/test_migrations.py

@@ -540,6 +540,9 @@ class SaharaMigrationsCheckers(object):
         self.assertColumnExists(engine, 'job_executions',
                                 'engine_job_id')
 
+    def _check_028(self, engine, data):
+        self.assertColumnExists(engine, 'instances', 'storage_devices_number')
+
 
 class TestMigrationsMySQL(SaharaMigrationsCheckers,
                           base.BaseWalkMigrationTestCase,

sahara/tests/unit/plugins/ambari/test_configs.py

@@ -30,8 +30,10 @@ class AmbariConfigsTestCase(base.SaharaTestCase):
         self.ng.node_configs = {}
         self.ng.cluster = mock.Mock()
         self.ng.cluster.hadoop_version = "2.2"
-        self.ng.storage_paths = mock.Mock()
+        self.instance = mock.Mock()
-        self.ng.storage_paths.return_value = ["/data1", "/data2"]
+        self.instance.node_group = self.ng
+        self.instance.storage_paths = mock.Mock()
+        self.instance.storage_paths.return_value = ["/data1", "/data2"]
 
     def assertConfigEqual(self, expected, actual):
         self.assertEqual(len(expected), len(actual))
@@ -45,8 +47,8 @@ class AmbariConfigsTestCase(base.SaharaTestCase):
         self.assertEqual(len(expected), len(cnt_ex))
         self.assertEqual(len(actual), len(cnt_act))
 
-    def test_get_ng_params_default(self):
+    def test_get_instance_params_default(self):
-        ng_configs = configs.get_ng_params(self.ng)
+        instance_configs = configs.get_instance_params(self.instance)
         expected = [
             {
                 "hdfs-site": {
@@ -71,16 +73,16 @@ class AmbariConfigsTestCase(base.SaharaTestCase):
                 }
             }
         ]
-        self.assertConfigEqual(expected, ng_configs)
+        self.assertConfigEqual(expected, instance_configs)
 
-    def test_get_ng_params(self):
+    def test_get_instance_params(self):
         self.ng.node_configs = {
             "YARN": {
                 "mapreduce.map.java.opts": "-Dk=v",
                 "yarn.scheduler.minimum-allocation-mb": "256"
             }
         }
-        ng_configs = configs.get_ng_params(self.ng)
+        instance_configs = configs.get_instance_params(self.instance)
         expected = [
             {
                 "hdfs-site": {
@@ -111,4 +113,4 @@ class AmbariConfigsTestCase(base.SaharaTestCase):
                 }
             }
         ]
-        self.assertConfigEqual(expected, ng_configs)
+        self.assertConfigEqual(expected, instance_configs)

sahara/tests/unit/plugins/hdp/hdp_test_base.py

@@ -29,6 +29,11 @@ class TestServer(object):
         self.public_ip = public_ip
         self.internal_ip = private_ip
         self.node_group = None
+        self.sahara_instance = self
+        self.storage_path = ['/mnt']
+
+    def storage_paths(self):
+        return self.storage_path
 
     def fqdn(self):
         return self.inst_fqdn
@@ -81,10 +86,6 @@ class TestNodeGroup(object):
         self.node_processes = node_processes
         self.count = count
         self.id = name
-        self.ng_storage_paths = []
-
-    def storage_paths(self):
-        return self.ng_storage_paths
 
 
 class TestUserInputConfig(object):

sahara/tests/unit/plugins/hdp/test_services.py

@@ -763,25 +763,32 @@ class ServicesTest(base.SaharaTestCase):
         for version in versions:
             s = self.get_services_processor(version=version)
             service = s.create_service('AMBARI')
-            ng1 = hdp_test_base.TestNodeGroup(None, None, None)
+            server1 = hdp_test_base.TestServer(
-            ng1.ng_storage_paths = ['/mnt', '/volume/disk1']
+                'host1', 'test-master', '11111', 3, '1.1.1.1', '2.2.2.2')
-            ng2 = hdp_test_base.TestNodeGroup(None, None, None)
+            server2 = hdp_test_base.TestServer(
-            ng2.ng_storage_paths = ['/mnt']
+                'host2', 'test-slave', '11111', 3, '3.3.3.3', '4.4.4.4')
+            server3 = hdp_test_base.TestServer(
+                'host3', 'another-test', '11111', 3, '6.6.6.6', '5.5.5.5')
+            ng1 = hdp_test_base.TestNodeGroup('ng1', [server1], None)
+            ng2 = hdp_test_base.TestNodeGroup('ng2', [server2], None)
+            ng3 = hdp_test_base.TestNodeGroup('ng3', [server3], None)
+
+            server1.storage_path = ['/volume/disk1']
+            server2.storage_path = ['/mnt']
 
             paths = service._get_common_paths([ng1, ng2])
-            self.assertEqual(['/mnt'], paths)
+            self.assertEqual([], paths)
 
-            ng3 = hdp_test_base.TestNodeGroup(None, None, None)
+            server1.storage_path = ['/volume/disk1', '/volume/disk2']
-            ng1.ng_storage_paths = ['/mnt', '/volume/disk1', '/volume/disk2']
+            server2.storage_path = ['/mnt']
-            ng2.ng_storage_paths = ['/mnt']
+            server3.storage_path = ['/volume/disk1']
-            ng3.ng_storage_paths = ['/mnt', '/volume/disk1']
 
             paths = service._get_common_paths([ng1, ng2, ng3])
-            self.assertEqual(['/mnt'], paths)
+            self.assertEqual([], paths)
 
-            ng1.ng_storage_paths = ['/mnt', '/volume/disk1', '/volume/disk2']
+            server1.storage_path = ['/volume/disk1', '/volume/disk2']
-            ng2.ng_storage_paths = ['/mnt', '/volume/disk1']
+            server2.storage_path = ['/volume/disk1']
-            ng3.ng_storage_paths = ['/mnt', '/volume/disk1']
+            server3.storage_path = ['/volume/disk1']
 
             paths = service._get_common_paths([ng1, ng2, ng3])
             self.assertEqual(['/volume/disk1'], paths)

sahara/tests/unit/service/test_volumes.py

@@ -79,13 +79,13 @@ class TestAttachVolume(base.SaharaWithDbTestCase):
             volumes.detach_from_instance(instance))
 
     @base.mock_thread_group
-    @mock.patch('sahara.service.volumes._mount_volume')
+    @mock.patch('sahara.service.volumes.mount_to_instances')
     @mock.patch('sahara.service.volumes._await_attach_volumes')
     @mock.patch('sahara.service.volumes._create_attach_volume')
     @mock.patch('sahara.utils.cluster_progress_ops.add_successful_event')
     @mock.patch('sahara.utils.cluster_progress_ops.add_provisioning_step')
-    def test_attach(self, add_step, add_event,
+    def test_attach(self, add_step, add_event, p_create_attach_vol, p_await,
-                    p_create_attach_vol, p_await, p_mount):
+                    p_mount):
         p_create_attach_vol.side_effect = ['/dev/vdb', '/dev/vdc'] * 2
         p_await.return_value = None
         p_mount.return_value = None
@@ -115,7 +115,7 @@ class TestAttachVolume(base.SaharaWithDbTestCase):
         volumes.attach_to_instances(cluster_utils.get_instances(cluster))
         self.assertEqual(4, p_create_attach_vol.call_count)
         self.assertEqual(2, p_await.call_count)
-        self.assertEqual(4, p_mount.call_count)
+        self.assertEqual(1, p_mount.call_count)
 
     @mock.patch('sahara.utils.poll_utils._get_consumed', return_value=0)
     @mock.patch('sahara.context.sleep')
@@ -157,3 +157,35 @@ class TestAttachVolume(base.SaharaWithDbTestCase):
         inst.remote.return_value = inst_remote
 
         return inst
+
+    def test_find_instance_volume_devices(self):
+        instance = self._get_instance()
+        ex_cmd = instance.remote().execute_command
+
+        attached_info = '/dev/vda /dev/vda1 /dev/vdb /dev/vdc'
+        mounted_info = '/dev/vda1'
+        ex_cmd.side_effect = [(0, attached_info), (0, mounted_info)]
+
+        diff = volumes._find_instance_devices(instance)
+        self.assertEqual(['/dev/vdb', '/dev/vdc'], diff)
+
+        attached_info = '/dev/vda /dev/vda1 /dev/vdb /dev/vdb1 /dev/vdb2'
+        mounted_info = '/dev/vda1'
+        ex_cmd.side_effect = [(0, attached_info), (0, mounted_info)]
+
+        diff = volumes._find_instance_devices(instance)
+        self.assertEqual(['/dev/vdb'], diff)
+
+        attached_info = '/dev/vda /dev/vda1 /dev/vdb /dev/vdb1 /dev/vdb2'
+        mounted_info = '/dev/vda1 /dev/vdb1'
+        ex_cmd.side_effect = [(0, attached_info), (0, mounted_info)]
+
+        diff = volumes._find_instance_devices(instance)
+        self.assertEqual(['/dev/vdb2'], diff)
+
+        attached_info = '/dev/vda /dev/vda1 /dev/vdb /dev/vdb1 /dev/vdb2'
+        mounted_info = '/dev/vda1 /dev/vdb2'
+        ex_cmd.side_effect = [(0, attached_info), (0, mounted_info)]
+
+        diff = volumes._find_instance_devices(instance)
+        self.assertEqual(['/dev/vdb1'], diff)