Add cluster validation to vanilla 2 plugin

Partially implements blueprint: vanilla-plugin-hadoop-2

Change-Id: Ib90a425366f7a4d7caaedf6cf8e3994c41a4e160
This commit is contained in:
Sergey Reshetnyak 2014-02-27 11:38:01 +04:00
parent fa72234b48
commit 698c937f93
4 changed files with 155 additions and 2 deletions

View File

@ -0,0 +1,85 @@
# Copyright (c) 2014 Mirantis Inc.
#
# 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.
from savanna.plugins.general import exceptions as ex
from savanna.plugins.general import utils as u
from savanna.plugins.vanilla.v2_3_0 import config_helper as c_helper
from savanna.utils import general as gu
def validate_cluster_creating(cluster):
nn_count = _get_inst_count(cluster, 'namenode')
if nn_count != 1:
raise ex.InvalidComponentCountException('namenode', 1, nn_count)
rm_count = _get_inst_count(cluster, 'resourcemanager')
if rm_count not in [0, 1]:
raise ex.InvalidComponentCountException('resourcemanager', '0 or 1',
rm_count)
if rm_count == 0:
nm_count = _get_inst_count(cluster, 'nodemanager')
if nm_count > 0:
raise ex.RequiredServiceMissingException('resourcemanager',
required_by='nodemanager')
def validate_additional_ng_scaling(cluster, additional):
rm = u.get_resourcemanager(cluster)
scalable_processes = _get_scalable_processes()
for ng_id in additional:
ng = gu.get_by_id(cluster.node_groups, ng_id)
if not set(ng.node_processes).issubset(scalable_processes):
msg = "Vanilla plugin cannot scale nodegroup with processes: %s"
raise ex.NodeGroupCannotBeScaled(ng.name,
msg % ' '.join(ng.node_processes))
if not rm and 'nodemanager' in ng.node_processes:
msg = ("Vanilla plugin cannot scale node group with processes "
"which have no master-processes run in cluster")
raise ex.NodeGroupCannotBeScaled(ng.name, msg)
def validate_existing_ng_scaling(cluster, existing):
scalable_processes = _get_scalable_processes()
dn_to_delete = 0
for ng in cluster.node_groups:
if ng.id in existing:
if ng.count > existing[ng.id] and "datanode" in ng.node_processes:
dn_to_delete += ng.count - existing[ng.id]
if not set(ng.node_processes).issubset(scalable_processes):
msg = ("Vanilla plugin cannot scale nodegroup "
"with processes: %s")
raise ex.NodeGroupCannotBeScaled(
ng.name, msg % ' '.join(ng.node_processes))
dn_amount = len(u.get_datanodes(cluster))
rep_factor = c_helper.get_config_value('HDFS', 'dfs.replication', cluster)
if dn_to_delete > 0 and dn_amount - dn_to_delete < rep_factor:
msg = ("Vanilla plugin cannot shrink cluster because it would be not "
"enough nodes for replicas (replication factor is %s)")
raise ex.ClusterCannotBeScaled(
cluster.name, msg % rep_factor)
def _get_scalable_processes():
return ['datanode', 'nodemanager']
def _get_inst_count(cluster, process):
return sum([ng.count for ng in u.get_node_groups(cluster, process)])

View File

@ -24,6 +24,7 @@ from savanna.plugins.vanilla.v2_3_0 import config as c
from savanna.plugins.vanilla.v2_3_0 import config_helper as c_helper
from savanna.plugins.vanilla.v2_3_0 import run_scripts as run
from savanna.plugins.vanilla.v2_3_0 import scaling as sc
from savanna.plugins.vanilla.v2_3_0 import validation as vl
conductor = conductor.API
LOG = logging.getLogger(__name__)
@ -43,7 +44,7 @@ class VersionHandler(avm.AbstractVersionHandler):
}
def validate(self, cluster):
pass
vl.validate_cluster_creating(cluster)
def update_infra(self, cluster):
pass
@ -71,7 +72,8 @@ class VersionHandler(avm.AbstractVersionHandler):
sc.decommission_nodes(cluster, instances)
def validate_scaling(self, cluster, existing, additional):
pass
vl.validate_additional_ng_scaling(cluster, additional)
vl.validate_existing_ng_scaling(cluster, existing)
def scale_cluster(self, cluster, instances):
sc.scale_cluster(cluster, instances)

View File

@ -0,0 +1,58 @@
# Copyright (c) 2014 Mirantis Inc.
#
# 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.
from savanna.plugins.general import exceptions as ex
from savanna.plugins.vanilla import plugin as p
from savanna.tests.unit import base
from savanna.tests.unit import testutils as tu
class ValidationTest(base.SavannaTestCase):
def setUp(self):
super(ValidationTest, self).setUp()
self.pl = p.VanillaProvider()
def test_validate(self):
self.ng = []
self.ng.append(tu.make_ng_dict("nn", "f1", ["namenode"], 0))
self.ng.append(tu.make_ng_dict("jt", "f1", ["resourcemanager"], 0))
self.ng.append(tu.make_ng_dict("tt", "f1", ["nodemanager"], 0))
self.ng.append(tu.make_ng_dict("dn", "f1", ["datanode"], 0))
self._validate_case(1, 1, 10, 10)
self._validate_case(1, 1, 1, 0)
self._validate_case(1, 1, 0, 1)
self._validate_case(1, 1, 0, 0)
self._validate_case(1, 0, 0, 0)
with self.assertRaises(ex.InvalidComponentCountException):
self._validate_case(0, 1, 10, 1)
with self.assertRaises(ex.InvalidComponentCountException):
self._validate_case(2, 1, 10, 1)
with self.assertRaises(ex.RequiredServiceMissingException):
self._validate_case(1, 0, 10, 1)
with self.assertRaises(ex.InvalidComponentCountException):
self._validate_case(1, 2, 10, 1)
def _validate_case(self, *args):
lst = []
for i in range(0, len(args)):
self.ng[i]['count'] = args[i]
lst.append(self.ng[i])
cl = tu.create_cluster("cluster1", "tenant1", "vanilla", "2.3.0", lst)
self.pl.validate(cl)

View File

@ -50,6 +50,14 @@ def find(lst, **kwargs):
return None
def get_by_id(lst, id):
for obj in lst:
if obj.id == id:
return obj
return None
def format_cluster_status(cluster):
msg = "Cluster status has been changed: id=%s, New status=%s"
if cluster: