<s> from oslo_log import log as logging <EOL> from oslo_serialization import jsonutils as json <EOL> from sahara . i18n import _ <EOL> from sahara . plugins import exceptions as ex <EOL> from sahara . plugins . hdp . versions import versionhandlerfactory as vhf <EOL> LOG = logging . getLogger ( __name__ ) <EOL> def validate_number_of_datanodes ( cluster , scaled_groups , default_configs ) : <EOL> dfs_replication = <NUM_LIT:0> <EOL> for config in default_configs : <EOL> if config . name == '<STR_LIT>' : <EOL> dfs_replication = config . default_value <EOL> conf = cluster . cluster_configs <EOL> if '<STR_LIT>' in conf and '<STR_LIT>' in conf [ '<STR_LIT>' ] : <EOL> dfs_replication = conf [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> if not scaled_groups : <EOL> scaled_groups = { } <EOL> dn_count = <NUM_LIT:0> <EOL> for ng in cluster . node_groups : <EOL> if '<STR_LIT>' in ng . node_processes : <EOL> if ng . id in scaled_groups : <EOL> dn_count += scaled_groups [ ng . id ] <EOL> else : <EOL> dn_count += ng . count <EOL> if dn_count < int ( dfs_replication ) : <EOL> raise ex . InvalidComponentCountException ( <EOL> '<STR_LIT>' , _ ( '<STR_LIT>' ) % dfs_replication , dn_count , <EOL> _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> % { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> class ClusterSpec ( object ) : <EOL> def __init__ ( self , config , version = '<STR_LIT>' ) : <EOL> self . _config_template = config <EOL> self . services = [ ] <EOL> self . configurations = { } <EOL> self . node_groups = { } <EOL> self . version = version <EOL> self . user_input_handlers = { } <EOL> cluster_template = json . loads ( config ) <EOL> self . _parse_services ( cluster_template ) <EOL> self . _parse_configurations ( cluster_template ) <EOL> self . _process_node_groups ( template_json = cluster_template ) <EOL> def create_operational_config ( self , cluster , user_inputs , <EOL> scaled_groups = None ) : <EOL> if scaled_groups is None : <EOL> scaled_groups = { } <EOL> self . _determine_deployed_services ( cluster ) <EOL> self . _process_node_groups ( cluster = cluster ) <EOL> for ng_id in scaled_groups : <EOL> existing = next ( group for group in self . node_groups . values ( ) <EOL> if group . id == ng_id ) <EOL> existing . count = scaled_groups [ ng_id ] <EOL> self . validate_node_groups ( cluster ) <EOL> self . _finalize_ng_components ( ) <EOL> self . _parse_configurations ( json . loads ( self . _config_template ) ) <EOL> self . _process_user_inputs ( user_inputs ) <EOL> self . _replace_config_tokens ( ) <EOL> def scale ( self , updated_groups ) : <EOL> for ng_id in updated_groups : <EOL> existing = next ( group for group in self . node_groups . values ( ) <EOL> if group . id == ng_id ) <EOL> existing . count = updated_groups [ ng_id ] <EOL> def validate_node_groups ( self , cluster ) : <EOL> for service in self . services : <EOL> if service . deployed : <EOL> service . validate ( self , cluster ) <EOL> elif service . is_mandatory ( ) : <EOL> raise ex . RequiredServiceMissingException ( service . name ) <EOL> def get_deployed_configurations ( self ) : <EOL> configs = set ( ) <EOL> for service in self . services : <EOL> if service . deployed : <EOL> configs |= service . configurations <EOL> return configs <EOL> def determine_component_hosts ( self , component ) : <EOL> hosts = set ( ) <EOL> for ng in self . node_groups . values ( ) : <EOL> if component in ng . components : <EOL> hosts |= ng . instances <EOL> return hosts <EOL> def normalize ( self ) : <EOL> return NormalizedClusterConfig ( self ) <EOL> def get_deployed_node_group_count ( self , name ) : <EOL> count = <NUM_LIT:0> <EOL> for ng in self . get_node_groups_containing_component ( name ) : <EOL> count += ng . count <EOL> return count <EOL> def get_node_groups_containing_component ( self , component ) : <EOL> found_node_groups = [ ] <EOL> for ng in self . node_groups . values ( ) : <EOL> if component in ng . components : <EOL> found_node_groups . append ( ng ) <EOL> return found_node_groups <EOL> def get_components_for_type ( self , type ) : <EOL> components = set ( ) <EOL> for service in self . services : <EOL> for component in service . components : <EOL> if component . type == type : <EOL> components . add ( component . name ) <EOL> return components <EOL> def is_hdfs_ha_enabled ( self , cluster ) : <EOL> if self . version == '<STR_LIT>' : <EOL> if cluster . cluster_configs . get ( '<STR_LIT>' , False ) : <EOL> if cluster . cluster_configs . HDFSHA . get ( '<STR_LIT>' , <EOL> False ) is True : <EOL> return True <EOL> return False <EOL> def _parse_services ( self , template_json ) : <EOL> handler = ( vhf . VersionHandlerFactory . get_instance ( ) . <EOL> get_version_handler ( self . version ) ) <EOL> sp = handler . get_services_processor ( ) <EOL> for s in template_json [ '<STR_LIT>' ] : <EOL> name = s [ '<STR_LIT:name>' ] <EOL> service = sp . create_service ( name ) <EOL> self . services . append ( service ) <EOL> for c in s [ '<STR_LIT>' ] : <EOL> component = Component ( c [ '<STR_LIT:name>' ] , c [ '<STR_LIT:type>' ] , c [ '<STR_LIT>' ] ) <EOL> service . add_component ( component ) <EOL> if '<STR_LIT>' in s : <EOL> for u in s [ '<STR_LIT>' ] : <EOL> user = User ( u [ '<STR_LIT:name>' ] , u [ '<STR_LIT:password>' ] , u [ '<STR_LIT>' ] ) <EOL> service . add_user ( user ) <EOL> configs = self . _parse_configurations ( s ) <EOL> for config in configs : <EOL> service . add_configuration ( config ) <EOL> def _parse_configurations ( self , template_json ) : <EOL> config_names = [ ] <EOL> for config in template_json [ '<STR_LIT>' ] : <EOL> config_props = { } <EOL> name = config [ '<STR_LIT:name>' ] <EOL> config_names . append ( name ) <EOL> if name in self . configurations : <EOL> config_props = self . configurations [ name ] <EOL> else : <EOL> self . configurations [ name ] = config_props <EOL> if '<STR_LIT>' in config : <EOL> for prop in config [ '<STR_LIT>' ] : <EOL> config_props [ prop [ '<STR_LIT:name>' ] ] = prop [ '<STR_LIT:value>' ] <EOL> return config_names <EOL> def _process_node_groups ( self , template_json = None , cluster = None ) : <EOL> if template_json and not cluster : <EOL> for group in template_json [ '<STR_LIT>' ] : <EOL> node_group = NodeGroup ( group [ '<STR_LIT:name>' ] ) <EOL> for component in group [ '<STR_LIT>' ] : <EOL> node_group . add_component ( component [ '<STR_LIT:name>' ] ) <EOL> for host in group [ '<STR_LIT>' ] : <EOL> if '<STR_LIT>' in host : <EOL> node_group . predicate = host [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in host : <EOL> node_group . cardinality = host [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in host : <EOL> node_group . count = host [ '<STR_LIT>' ] <EOL> self . node_groups [ node_group . name ] = node_group <EOL> if cluster : <EOL> self . node_groups = { } <EOL> node_groups = cluster . node_groups <EOL> for ng in node_groups : <EOL> node_group = NodeGroup ( ng . name ) <EOL> node_group . count = ng . count <EOL> node_group . id = ng . id <EOL> node_group . components = ng . node_processes [ : ] <EOL> for instance in ng . instances : <EOL> node_group . instances . add ( Instance ( instance ) ) <EOL> self . node_groups [ node_group . name ] = node_group <EOL> def _determine_deployed_services ( self , cluster ) : <EOL> for ng in cluster . node_groups : <EOL> for service in self . services : <EOL> if service . deployed : <EOL> continue <EOL> for sc in service . components : <EOL> if sc . name in ng . node_processes : <EOL> service . deployed = True <EOL> service . register_user_input_handlers ( <EOL> self . user_input_handlers ) <EOL> break <EOL> def _process_user_inputs ( self , user_inputs ) : <EOL> for ui in user_inputs : <EOL> if not hasattr ( ui . config , '<STR_LIT>' ) : <EOL> continue <EOL> user_input_handler = self . user_input_handlers . get ( <EOL> '<STR_LIT>' . format ( ui . config . tag , ui . config . name ) , <EOL> self . _default_user_input_handler ) <EOL> user_input_handler ( ui , self . configurations ) <EOL> def _replace_config_tokens ( self ) : <EOL> for service in self . services : <EOL> if service . deployed : <EOL> service . finalize_configuration ( self ) <EOL> def _finalize_ng_components ( self ) : <EOL> for service in self . services : <EOL> if service . deployed : <EOL> service . finalize_ng_components ( self ) <EOL> def _default_user_input_handler ( self , user_input , configurations ) : <EOL> config_map = configurations [ user_input . config . tag ] <EOL> config_map [ user_input . config . name ] = user_input . value <EOL> class Component ( object ) : <EOL> def __init__ ( self , name , component_type , cardinality ) : <EOL> self . name = name <EOL> self . type = component_type <EOL> self . cardinality = cardinality <EOL> class NodeGroup ( object ) : <EOL> def __init__ ( self , name ) : <EOL> self . id = None <EOL> self . name = name <EOL> self . components = [ ] <EOL> self . predicate = None <EOL> self . cardinality = None <EOL> self . count = None <EOL> self . instances = set ( ) <EOL> def add_component ( self , component ) : <EOL> self . components . append ( component ) <EOL> class User ( object ) : <EOL> def __init__ ( self , name , password , groups ) : <EOL> self . name = name <EOL> self . password = password <EOL> self . groups = groups <EOL> class Instance ( object ) : <EOL> def __init__ ( self , sahara_instance ) : <EOL> self . inst_fqdn = sahara_instance . fqdn ( ) <EOL> self . management_ip = sahara_instance . management_ip <EOL> self . internal_ip = sahara_instance . internal_ip <EOL> self . sahara_instance = sahara_instance <EOL> def fqdn ( self ) : <EOL> return self . inst_fqdn <EOL> def remote ( self ) : <EOL> return self . sahara_instance . remote ( ) <EOL> def __hash__ ( self ) : <EOL> return hash ( self . fqdn ( ) ) <EOL> def __eq__ ( self , other ) : <EOL> return self . fqdn ( ) == other . fqdn ( ) <EOL> class NormalizedClusterConfig ( object ) : <EOL> def __init__ ( self , cluster_spec ) : <EOL> self . hadoop_version = cluster_spec . version <EOL> self . cluster_configs = [ ] <EOL> self . node_groups = [ ] <EOL> self . handler = ( vhf . VersionHandlerFactory . get_instance ( ) . <EOL> get_version_handler ( self . hadoop_version ) ) <EOL> self . _parse_configurations ( cluster_spec . configurations ) <EOL> self . _parse_node_groups ( cluster_spec . node_groups ) <EOL> def _parse_configurations ( self , configurations ) : <EOL> for config_name , properties in configurations . items ( ) : <EOL> for prop , value in properties . items ( ) : <EOL> target = self . _get_property_target ( prop ) <EOL> if target : <EOL> prop_type = self . _get_property_type ( prop , value ) <EOL> self . cluster_configs . append ( <EOL> NormalizedConfigEntry ( NormalizedConfig ( <EOL> prop , prop_type , value , target , '<STR_LIT>' ) , <EOL> value ) ) <EOL> def _parse_node_groups ( self , node_groups ) : <EOL> for node_group in node_groups . values ( ) : <EOL> self . node_groups . append ( NormalizedNodeGroup ( node_group ) ) <EOL> def _get_property_target ( self , prop ) : <EOL> return self . handler . get_applicable_target ( prop ) <EOL> def _get_property_type ( self , prop , value ) : <EOL> prop_type = type ( value ) . __name__ <EOL> if prop_type == '<STR_LIT:str>' or prop_type == '<STR_LIT>' or value == '<STR_LIT>' : <EOL> return '<STR_LIT:string>' <EOL> elif prop_type == '<STR_LIT:int>' : <EOL> return '<STR_LIT>' <EOL> elif prop_type == '<STR_LIT:bool>' : <EOL> return '<STR_LIT>' <EOL> else : <EOL> raise ValueError ( <EOL> _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) % <EOL> { "<STR_LIT>" : prop , "<STR_LIT:value>" : value } ) <EOL> class NormalizedConfig ( object ) : <EOL> def __init__ ( self , name , config_type , default_value , target , scope ) : <EOL> self . name = name <EOL> self . description = None <EOL> self . type = config_type <EOL> self . default_value = default_value <EOL> self . is_optional = False <EOL> self . applicable_target = target <EOL> self . scope = scope <EOL> class NormalizedConfigEntry ( object ) : <EOL> def __init__ ( self , config , value ) : <EOL> self . config = config <EOL> self . value = value <EOL> class NormalizedNodeGroup ( object ) : <EOL> def __init__ ( self , node_group ) : <EOL> self . name = node_group . name <EOL> self . node_processes = node_group . components <EOL> self . node_configs = None <EOL> self . img = None <EOL> self . flavor = <NUM_LIT:3> <EOL> self . count = node_group . count <EOL> self . id = node_group . id </s>
