Merge "[HDP] Add validation check for dfs.replication"

sahara/plugins/hdp/clusterspec.py

@@ -24,6 +24,33 @@ from sahara.plugins.hdp.versions import versionhandlerfactory as vhf
 LOG = logging.getLogger(__name__)
 
 
+def validate_number_of_datanodes(cluster, scaled_groups, default_configs):
+    dfs_replication = 0
+    for config in default_configs:
+        if config.name == 'dfs.replication':
+            dfs_replication = config.default_value
+    conf = cluster.cluster_configs
+    if 'HDFS' in conf and 'dfs.replication' in conf['HDFS']:
+        dfs_replication = conf['HDFS']['dfs.replication']
+
+    if not scaled_groups:
+        scaled_groups = {}
+    dn_count = 0
+    for ng in cluster.node_groups:
+        if 'DATANODE' in ng.node_processes:
+            if ng.id in scaled_groups:
+                dn_count += scaled_groups[ng.id]
+            else:
+                dn_count += ng.count
+
+    if dn_count < int(dfs_replication):
+        raise ex.InvalidComponentCountException(
+            'datanode', _('%s or more') % dfs_replication, dn_count,
+            _('Number of %(dn)s instances should not be less '
+              'than %(replication)s')
+            % {'dn': 'DATANODE', 'replication': 'dfs.replication'})
+
+
 class ClusterSpec(object):
     def __init__(self, config, version='1.3.2'):
         self._config_template = config

sahara/plugins/hdp/versions/version_1_3_2/versionhandler.py

@@ -100,6 +100,9 @@ class VersionHandler(avm.AbstractVersionHandler):
             cluster_spec.create_operational_config(
                 cluster, user_inputs, scaled_groups)
 
+            cs.validate_number_of_datanodes(
+                cluster, scaled_groups, self.get_config_items())
+
         return cluster_spec
 
     def get_default_cluster_configuration(self):

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

@@ -90,6 +90,9 @@ class VersionHandler(avm.AbstractVersionHandler):
             cluster_spec.create_operational_config(
                 cluster, user_inputs, scaled_groups)
 
+            cs.validate_number_of_datanodes(
+                cluster, scaled_groups, self.get_config_items())
+
         return cluster_spec
 
     def get_default_cluster_configuration(self):