<s> from oslo_config import cfg <EOL> from oslo_log import log as logging <EOL> import six <EOL> from sahara import conductor as c <EOL> from sahara . plugins import provisioning as p <EOL> from sahara . plugins import utils <EOL> from sahara . swift import swift_helper as swift <EOL> from sahara . topology import topology_helper as topology <EOL> from sahara . utils import files as f <EOL> from sahara . utils import types as types <EOL> from sahara . utils import xmlutils as x <EOL> conductor = c . API <EOL> LOG = logging . getLogger ( __name__ ) <EOL> CONF = cfg . CONF <EOL> CORE_DEFAULT = x . load_hadoop_xml_defaults ( <EOL> '<STR_LIT>' ) <EOL> HDFS_DEFAULT = x . load_hadoop_xml_defaults ( <EOL> '<STR_LIT>' ) <EOL> SWIFT_DEFAULTS = swift . read_default_swift_configs ( ) <EOL> XML_CONFS = { <EOL> "<STR_LIT>" : [ CORE_DEFAULT , HDFS_DEFAULT , SWIFT_DEFAULTS ] <EOL> } <EOL> _default_executor_classpath = "<STR_LIT::>" . join ( <EOL> [ '<STR_LIT>' ] ) <EOL> SPARK_CONFS = { <EOL> '<STR_LIT>' : { <EOL> "<STR_LIT>" : [ <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % _default_executor_classpath , <EOL> '<STR_LIT:default>' : '<STR_LIT:%s>' % _default_executor_classpath , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:default>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:default>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT:default>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT:default>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:default>' : '<STR_LIT:all>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:default>' : '<STR_LIT:all>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:default>' : '<STR_LIT:1>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:default>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:default>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:default>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:default>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } , <EOL> ] <EOL> } <EOL> } <EOL> HADOOP_CONF_DIR = "<STR_LIT>" <EOL> ENV_CONFS = { <EOL> "<STR_LIT>" : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> } <EOL> ENABLE_DATA_LOCALITY = p . Config ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> config_type = "<STR_LIT:bool>" , priority = <NUM_LIT:1> , <EOL> default_value = True , is_optional = True ) <EOL> ENABLE_SWIFT = p . Config ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> config_type = "<STR_LIT:bool>" , priority = <NUM_LIT:1> , <EOL> default_value = True , is_optional = False ) <EOL> DATANODES_STARTUP_TIMEOUT = p . Config ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , config_type = '<STR_LIT:int>' , <EOL> priority = <NUM_LIT:1> , default_value = <NUM_LIT> , is_optional = True , <EOL> description = '<STR_LIT>' ) <EOL> DECOMMISSIONING_TIMEOUT = p . Config ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , config_type = '<STR_LIT:int>' , priority = <NUM_LIT:1> , <EOL> default_value = <NUM_LIT> , is_optional = True , <EOL> description = '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> HIDDEN_CONFS = [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> CLUSTER_WIDE_CONFS = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> PRIORITY_1_CONFS = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> PRIORITY_1_CONFS += CLUSTER_WIDE_CONFS <EOL> def _initialise_configs ( ) : <EOL> configs = [ ] <EOL> for service , config_lists in six . iteritems ( XML_CONFS ) : <EOL> for config_list in config_lists : <EOL> for config in config_list : <EOL> if config [ '<STR_LIT:name>' ] not in HIDDEN_CONFS : <EOL> cfg = p . Config ( config [ '<STR_LIT:name>' ] , service , "<STR_LIT>" , <EOL> is_optional = True , config_type = "<STR_LIT:string>" , <EOL> default_value = str ( config [ '<STR_LIT:value>' ] ) , <EOL> description = config [ '<STR_LIT:description>' ] ) <EOL> if cfg . default_value in [ "<STR_LIT:true>" , "<STR_LIT:false>" ] : <EOL> cfg . config_type = "<STR_LIT:bool>" <EOL> cfg . default_value = ( cfg . default_value == '<STR_LIT:true>' ) <EOL> elif types . is_int ( cfg . default_value ) : <EOL> cfg . config_type = "<STR_LIT:int>" <EOL> cfg . default_value = int ( cfg . default_value ) <EOL> if config [ '<STR_LIT:name>' ] in CLUSTER_WIDE_CONFS : <EOL> cfg . scope = '<STR_LIT>' <EOL> if config [ '<STR_LIT:name>' ] in PRIORITY_1_CONFS : <EOL> cfg . priority = <NUM_LIT:1> <EOL> configs . append ( cfg ) <EOL> for service , config_items in six . iteritems ( ENV_CONFS ) : <EOL> for name , param_format_str in six . iteritems ( config_items ) : <EOL> configs . append ( p . Config ( name , service , "<STR_LIT>" , <EOL> default_value = <NUM_LIT> , priority = <NUM_LIT:1> , <EOL> config_type = "<STR_LIT:int>" ) ) <EOL> for service , config_items in six . iteritems ( SPARK_CONFS ) : <EOL> for item in config_items [ '<STR_LIT>' ] : <EOL> cfg = p . Config ( name = item [ "<STR_LIT:name>" ] , <EOL> description = item [ "<STR_LIT:description>" ] , <EOL> default_value = item [ "<STR_LIT:default>" ] , <EOL> applicable_target = service , <EOL> scope = "<STR_LIT>" , is_optional = True , <EOL> priority = item [ "<STR_LIT>" ] ) <EOL> configs . append ( cfg ) <EOL> configs . append ( DECOMMISSIONING_TIMEOUT ) <EOL> configs . append ( ENABLE_SWIFT ) <EOL> configs . append ( DATANODES_STARTUP_TIMEOUT ) <EOL> if CONF . enable_data_locality : <EOL> configs . append ( ENABLE_DATA_LOCALITY ) <EOL> return configs <EOL> PLUGIN_CONFIGS = _initialise_configs ( ) <EOL> def get_plugin_configs ( ) : <EOL> return PLUGIN_CONFIGS <EOL> def generate_cfg_from_general ( cfg , configs , general_config , <EOL> rest_excluded = False ) : <EOL> if '<STR_LIT>' in configs : <EOL> for nm in general_config : <EOL> if nm not in configs [ '<STR_LIT>' ] and not rest_excluded : <EOL> configs [ '<STR_LIT>' ] [ nm ] = general_config [ nm ] [ '<STR_LIT>' ] <EOL> for name , value in configs [ '<STR_LIT>' ] . items ( ) : <EOL> if value : <EOL> cfg = _set_config ( cfg , general_config , name ) <EOL> LOG . debug ( "<STR_LIT>" . format ( name = name ) ) <EOL> else : <EOL> cfg = _set_config ( cfg , general_config ) <EOL> return cfg <EOL> def _get_hostname ( service ) : <EOL> return service . hostname ( ) if service else None <EOL> def generate_xml_configs ( configs , storage_path , nn_hostname , hadoop_port ) : <EOL> if hadoop_port is None : <EOL> hadoop_port = <NUM_LIT> <EOL> cfg = { <EOL> '<STR_LIT>' : '<STR_LIT>' % ( nn_hostname , str ( hadoop_port ) ) , <EOL> '<STR_LIT>' : extract_hadoop_path ( storage_path , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' : extract_hadoop_path ( storage_path , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> for key , value in extract_hadoop_xml_confs ( configs ) : <EOL> cfg [ key ] = value <EOL> swft_def = [ ] <EOL> if is_swift_enabled ( configs ) : <EOL> swft_def = SWIFT_DEFAULTS <EOL> swift_configs = extract_name_values ( swift . get_swift_configs ( ) ) <EOL> for key , value in six . iteritems ( swift_configs ) : <EOL> if key not in cfg : <EOL> cfg [ key ] = value <EOL> core_all = CORE_DEFAULT + swft_def <EOL> if CONF . enable_data_locality : <EOL> cfg . update ( topology . TOPOLOGY_CONFIG ) <EOL> core_all += topology . vm_awareness_core_config ( ) <EOL> xml_configs = { <EOL> '<STR_LIT>' : x . create_hadoop_xml ( cfg , core_all ) , <EOL> '<STR_LIT>' : x . create_hadoop_xml ( cfg , HDFS_DEFAULT ) <EOL> } <EOL> return xml_configs <EOL> def _get_spark_opt_default ( opt_name ) : <EOL> for opt in SPARK_CONFS [ "<STR_LIT>" ] [ "<STR_LIT>" ] : <EOL> if opt_name == opt [ "<STR_LIT:name>" ] : <EOL> return opt [ "<STR_LIT:default>" ] <EOL> return None <EOL> def generate_spark_env_configs ( cluster ) : <EOL> configs = [ ] <EOL> sp_master = utils . get_instance ( cluster , "<STR_LIT>" ) <EOL> configs . append ( '<STR_LIT>' + sp_master . hostname ( ) ) <EOL> configs . append ( '<STR_LIT>' + HADOOP_CONF_DIR ) <EOL> masterport = utils . get_config_value_or_default ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> cluster ) <EOL> if masterport and masterport != _get_spark_opt_default ( "<STR_LIT>" ) : <EOL> configs . append ( '<STR_LIT>' + str ( masterport ) ) <EOL> masterwebport = utils . get_config_value_or_default ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> cluster ) <EOL> if ( masterwebport and <EOL> masterwebport != _get_spark_opt_default ( "<STR_LIT>" ) ) : <EOL> configs . append ( '<STR_LIT>' + str ( masterwebport ) ) <EOL> workercores = utils . get_config_value_or_default ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> cluster ) <EOL> if workercores and workercores != _get_spark_opt_default ( "<STR_LIT>" ) : <EOL> configs . append ( '<STR_LIT>' + str ( workercores ) ) <EOL> workermemory = utils . get_config_value_or_default ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> cluster ) <EOL> if ( workermemory and <EOL> workermemory != _get_spark_opt_default ( "<STR_LIT>" ) ) : <EOL> configs . append ( '<STR_LIT>' + str ( workermemory ) ) <EOL> workerport = utils . get_config_value_or_default ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> cluster ) <EOL> if workerport and workerport != _get_spark_opt_default ( "<STR_LIT>" ) : <EOL> configs . append ( '<STR_LIT>' + str ( workerport ) ) <EOL> workerwebport = utils . get_config_value_or_default ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> cluster ) <EOL> if ( workerwebport and <EOL> workerwebport != _get_spark_opt_default ( "<STR_LIT>" ) ) : <EOL> configs . append ( '<STR_LIT>' + str ( workerwebport ) ) <EOL> workerinstances = utils . get_config_value_or_default ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> cluster ) <EOL> if ( workerinstances and <EOL> workerinstances != _get_spark_opt_default ( "<STR_LIT>" ) ) : <EOL> configs . append ( '<STR_LIT>' + str ( workerinstances ) ) <EOL> return '<STR_LIT:\n>' . join ( configs ) <EOL> def generate_spark_slaves_configs ( workernames ) : <EOL> return '<STR_LIT:\n>' . join ( workernames ) <EOL> def generate_spark_executor_classpath ( cluster ) : <EOL> cp = utils . get_config_value_or_default ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> cluster ) <EOL> if cp : <EOL> return "<STR_LIT>" + cp <EOL> return "<STR_LIT:\n>" <EOL> def extract_hadoop_environment_confs ( configs ) : <EOL> """<STR_LIT>""" <EOL> lst = [ ] <EOL> for service , srv_confs in configs . items ( ) : <EOL> if ENV_CONFS . get ( service ) : <EOL> for param_name , param_value in srv_confs . items ( ) : <EOL> for cfg_name , cfg_format_str in ENV_CONFS [ service ] . items ( ) : <EOL> if param_name == cfg_name and param_value is not None : <EOL> lst . append ( cfg_format_str % param_value ) <EOL> return lst <EOL> def extract_hadoop_xml_confs ( configs ) : <EOL> """<STR_LIT>""" <EOL> lst = [ ] <EOL> for service , srv_confs in configs . items ( ) : <EOL> if XML_CONFS . get ( service ) : <EOL> for param_name , param_value in srv_confs . items ( ) : <EOL> for cfg_list in XML_CONFS [ service ] : <EOL> names = [ cfg [ '<STR_LIT:name>' ] for cfg in cfg_list ] <EOL> if param_name in names and param_value is not None : <EOL> lst . append ( ( param_name , param_value ) ) <EOL> return lst <EOL> def generate_hadoop_setup_script ( storage_paths , env_configs ) : <EOL> script_lines = [ "<STR_LIT>" ] <EOL> script_lines . append ( "<STR_LIT>" ) <EOL> for line in env_configs : <EOL> if '<STR_LIT>' in line : <EOL> script_lines . append ( '<STR_LIT>' % line ) <EOL> script_lines . append ( "<STR_LIT>" ) <EOL> script_lines . append ( "<STR_LIT>" ) <EOL> hadoop_log = storage_paths [ <NUM_LIT:0> ] + "<STR_LIT>" <EOL> script_lines . append ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % hadoop_log ) <EOL> hadoop_log = storage_paths [ <NUM_LIT:0> ] + "<STR_LIT>" <EOL> script_lines . append ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % hadoop_log ) <EOL> for path in storage_paths : <EOL> script_lines . append ( "<STR_LIT>" % path ) <EOL> script_lines . append ( "<STR_LIT>" <EOL> "<STR_LIT>" % path ) <EOL> return "<STR_LIT:\n>" . join ( script_lines ) <EOL> def generate_job_cleanup_config ( cluster ) : <EOL> args = { <EOL> '<STR_LIT>' : utils . get_config_value_or_default ( <EOL> "<STR_LIT>" , "<STR_LIT>" , cluster ) , <EOL> '<STR_LIT>' : utils . get_config_value_or_default ( <EOL> "<STR_LIT>" , "<STR_LIT>" , cluster ) , <EOL> '<STR_LIT>' : utils . get_config_value_or_default ( <EOL> "<STR_LIT>" , "<STR_LIT>" , cluster ) <EOL> } <EOL> job_conf = { '<STR_LIT>' : ( args [ '<STR_LIT>' ] > <NUM_LIT:0> and <EOL> ( args [ '<STR_LIT>' ] > <NUM_LIT:0> <EOL> and args [ '<STR_LIT>' ] > <NUM_LIT:0> ) ) } <EOL> if job_conf [ '<STR_LIT>' ] : <EOL> job_conf [ '<STR_LIT>' ] = f . get_file_text ( <EOL> '<STR_LIT>' ) , <EOL> job_cleanup_script = f . get_file_text ( <EOL> '<STR_LIT>' ) <EOL> job_conf [ '<STR_LIT>' ] = job_cleanup_script . format ( ** args ) <EOL> return job_conf <EOL> def extract_name_values ( configs ) : <EOL> return { cfg [ '<STR_LIT:name>' ] : cfg [ '<STR_LIT:value>' ] for cfg in configs } <EOL> def make_hadoop_path ( base_dirs , suffix ) : <EOL> return [ base_dir + suffix for base_dir in base_dirs ] <EOL> def extract_hadoop_path ( lst , hadoop_dir ) : <EOL> if lst : <EOL> return "<STR_LIT:U+002C>" . join ( make_hadoop_path ( lst , hadoop_dir ) ) <EOL> def _set_config ( cfg , gen_cfg , name = None ) : <EOL> if name in gen_cfg : <EOL> cfg . update ( gen_cfg [ name ] [ '<STR_LIT>' ] ) <EOL> if name is None : <EOL> for name in gen_cfg : <EOL> cfg . update ( gen_cfg [ name ] [ '<STR_LIT>' ] ) <EOL> return cfg <EOL> def _get_general_config_value ( conf , option ) : <EOL> if '<STR_LIT>' in conf and option . name in conf [ '<STR_LIT>' ] : <EOL> return conf [ '<STR_LIT>' ] [ option . name ] <EOL> return option . default_value <EOL> def _get_general_cluster_config_value ( cluster , option ) : <EOL> return _get_general_config_value ( cluster . cluster_configs , option ) <EOL> def is_data_locality_enabled ( cluster ) : <EOL> if not CONF . enable_data_locality : <EOL> return False <EOL> return _get_general_cluster_config_value ( cluster , ENABLE_DATA_LOCALITY ) <EOL> def is_swift_enabled ( configs ) : <EOL> return _get_general_config_value ( configs , ENABLE_SWIFT ) <EOL> def get_decommissioning_timeout ( cluster ) : <EOL> return _get_general_cluster_config_value ( cluster , DECOMMISSIONING_TIMEOUT ) <EOL> def get_port_from_config ( service , name , cluster = None ) : <EOL> address = utils . get_config_value_or_default ( service , name , cluster ) <EOL> return utils . get_port_from_address ( address ) </s>
