<s> try : <EOL> from sshtunnel import SSHTunnelForwarder <EOL> except ImportError : <EOL> from sshtunnel . sshtunnel import SSHTunnelForwarder <EOL> from cm_api . api_client import ApiResource , ApiException <EOL> from cm_api . endpoints . services import ApiService , ApiServiceSetupInfo <EOL> import paramiko <EOL> import json <EOL> import yaml <EOL> import requests <EOL> import subprocess <EOL> import zipfile <EOL> import shutil <EOL> import os <EOL> import logger <EOL> import base64 <EOL> class CdhConfExtractor ( object ) : <EOL> def __init__ ( self , config_filename = None ) : <EOL> self . _logger = logger . get_info_logger ( __name__ ) <EOL> self . config_filename = config_filename if config_filename else '<STR_LIT>' <EOL> config = self . _load_config_yaml ( self . config_filename ) <EOL> self . _hostname = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . _hostport = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . _username = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:username>' ] <EOL> self . _key_filename = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . _key = os . path . expanduser ( self . _key_filename ) <EOL> self . _key_password = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . _is_openstack = config [ '<STR_LIT>' ] <EOL> self . _is_kerberos = config [ '<STR_LIT>' ] <EOL> self . _cdh_manager_ip = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . _cdh_manager_user = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:user>' ] <EOL> self . _cdh_manager_sshtunnel_required = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . _cdh_manager_password = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:password>' ] <EOL> def __enter__ ( self ) : <EOL> extractor = self <EOL> try : <EOL> if self . _cdh_manager_sshtunnel_required : <EOL> self . _logger . info ( '<STR_LIT>' ) <EOL> extractor . create_tunnel_to_cdh_manager ( ) <EOL> extractor . start_cdh_manager_tunneling ( ) <EOL> self . _logger . info ( '<STR_LIT>' ) <EOL> else : <EOL> self . _logger . info ( '<STR_LIT>' ) <EOL> self . _local_bind_address = self . extract_cdh_manager_host ( ) <EOL> self . _local_bind_port = <NUM_LIT> <EOL> return extractor <EOL> except Exception as exc : <EOL> self . _logger . error ( '<STR_LIT>' ) <EOL> raise exc <EOL> def __exit__ ( self , exc_type , exc_val , exc_tb ) : <EOL> try : <EOL> if self . _cdh_manager_sshtunnel_required : <EOL> self . stop_cdh_manager_tunneling ( ) <EOL> self . _logger . info ( '<STR_LIT>' ) <EOL> except Exception as exc : <EOL> self . _logger . error ( '<STR_LIT>' ) <EOL> raise exc <EOL> def create_ssh_connection ( self , hostname , username , key_filename , key_password ) : <EOL> try : <EOL> self . _logger . info ( '<STR_LIT>' . format ( hostname ) ) <EOL> self . ssh_connection = paramiko . SSHClient ( ) <EOL> self . ssh_connection . set_missing_host_key_policy ( paramiko . AutoAddPolicy ( ) ) <EOL> self . ssh_connection . connect ( hostname , username = username , key_filename = key_filename , password = key_password ) <EOL> self . _logger . info ( '<STR_LIT>' . format ( hostname ) ) <EOL> except Exception as exc : <EOL> self . _logger . error ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( hostname ) ) <EOL> raise exc <EOL> def close_ssh_connection ( self ) : <EOL> try : <EOL> self . ssh_connection . close ( ) <EOL> self . _logger . info ( '<STR_LIT>' ) <EOL> except Exception as exc : <EOL> self . _logger . error ( '<STR_LIT>' ) <EOL> raise exc <EOL> def ssh_call_command ( self , command , subcommands = None ) : <EOL> self . _logger . info ( '<STR_LIT>' . format ( command , subcommands ) ) <EOL> ssh_in , ssh_out , ssh_err = self . ssh_connection . exec_command ( command , get_pty = True ) <EOL> if subcommands != None : <EOL> for subcommand in subcommands : <EOL> ssh_in . write ( subcommand + '<STR_LIT:\n>' ) <EOL> ssh_in . flush ( ) <EOL> return ssh_out . read ( ) if ssh_out is not None else ssh_err . read ( ) <EOL> def extract_cdh_manager_host ( self ) : <EOL> self . _logger . info ( '<STR_LIT>' ) <EOL> if self . _cdh_manager_ip is None : <EOL> self . create_ssh_connection ( self . _hostname , self . _username , self . _key_filename , self . _key_password ) <EOL> if self . _is_openstack : <EOL> ansible_ini = self . ssh_call_command ( '<STR_LIT>' ) <EOL> else : <EOL> ansible_ini = self . ssh_call_command ( '<STR_LIT>' ) <EOL> self . _cdh_manager_ip = self . _get_host_ip ( '<STR_LIT>' , ansible_ini ) <EOL> self . close_ssh_connection ( ) <EOL> self . _logger . info ( '<STR_LIT>' . format ( self . _cdh_manager_ip ) ) <EOL> return self . _cdh_manager_ip <EOL> def create_tunnel_to_cdh_manager ( self , local_bind_address = '<STR_LIT:localhost>' , local_bind_port = <NUM_LIT> , remote_bind_port = <NUM_LIT> ) : <EOL> self . _local_bind_address = local_bind_address <EOL> self . _local_bind_port = local_bind_port <EOL> self . cdh_manager_tunnel = SSHTunnelForwarder ( <EOL> ( self . _hostname , self . _hostport ) , <EOL> ssh_username = self . _username , <EOL> local_bind_address = ( local_bind_address , local_bind_port ) , <EOL> remote_bind_address = ( self . extract_cdh_manager_host ( ) , remote_bind_port ) , <EOL> ssh_private_key_password = self . _key_password , <EOL> ssh_private_key = self . _key <EOL> ) <EOL> def start_cdh_manager_tunneling ( self ) : <EOL> try : <EOL> self . cdh_manager_tunnel . start ( ) <EOL> except Exception as e : <EOL> self . _logger . error ( '<STR_LIT>' + e . message ) <EOL> def stop_cdh_manager_tunneling ( self ) : <EOL> try : <EOL> self . cdh_manager_tunnel . stop ( ) <EOL> except Exception as e : <EOL> self . _logger . error ( '<STR_LIT>' + e . message ) <EOL> def extract_cdh_manager_details ( self , settings ) : <EOL> for host in settings [ '<STR_LIT>' ] : <EOL> if '<STR_LIT>' in host [ '<STR_LIT>' ] : <EOL> return host <EOL> def extract_nodes_info ( self , name , settings ) : <EOL> nodes = [ ] <EOL> for host in settings [ '<STR_LIT>' ] : <EOL> if name in host [ '<STR_LIT>' ] : <EOL> nodes . append ( host ) <EOL> return nodes <EOL> def extract_service_namenode ( self , service_name , role_name , settings ) : <EOL> hdfs_service = self . _find_item_by_attr_value ( service_name , '<STR_LIT:name>' , settings [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> hdfs_namenode = self . _find_item_by_attr_value ( role_name , '<STR_LIT:name>' , hdfs_service [ '<STR_LIT>' ] ) <EOL> host_id = hdfs_namenode [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> return self . _find_item_by_attr_value ( host_id , '<STR_LIT>' , settings [ '<STR_LIT>' ] ) [ '<STR_LIT>' ] <EOL> def get_client_config_for_service ( self , service_name ) : <EOL> result = requests . get ( '<STR_LIT>' . format ( self . _local_bind_address , self . _local_bind_port , service_name ) ) <EOL> return base64 . standard_b64encode ( result . content ) <EOL> def generate_keytab ( self , principal_name ) : <EOL> self . _logger . info ( '<STR_LIT>' . format ( principal_name ) ) <EOL> self . create_ssh_connection ( self . _hostname , self . _username , self . _key_filename , self . _key_password ) <EOL> sftp = self . ssh_connection . open_sftp ( ) <EOL> sftp . put ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . ssh_call_command ( '<STR_LIT>' . format ( self . _cdh_manager_ip ) ) <EOL> self . ssh_call_command ( '<STR_LIT>' . format ( self . _cdh_manager_ip ) ) <EOL> keytab_hash = self . ssh_call_command ( '<STR_LIT>' <EOL> . format ( self . _cdh_manager_ip , principal_name ) ) <EOL> self . close_ssh_connection ( ) <EOL> lines = keytab_hash . splitlines ( ) <EOL> self . _logger . info ( '<STR_LIT>' . format ( principal_name ) ) <EOL> return '<STR_LIT>' . format ( '<STR_LIT>' . join ( lines [ <NUM_LIT:2> : - <NUM_LIT:2> ] ) ) <EOL> def generate_base64_for_file ( self , file_path , hostname ) : <EOL> self . _logger . info ( '<STR_LIT>' . format ( file_path ) ) <EOL> self . create_ssh_connection ( self . _hostname , self . _username , self . _key_filename , self . _key_password ) <EOL> base64_file_hash = self . ssh_call_command ( '<STR_LIT>' . format ( hostname , file_path ) ) <EOL> self . close_ssh_connection ( ) <EOL> lines = base64_file_hash . splitlines ( ) <EOL> self . _logger . info ( '<STR_LIT>' . format ( file_path , hostname ) ) <EOL> return '<STR_LIT>' . format ( '<STR_LIT>' . join ( lines [ <NUM_LIT:2> : - <NUM_LIT:2> ] ) ) <EOL> def get_all_deployments_conf ( self ) : <EOL> result = { } <EOL> deployments_settings = json . loads ( requests . get ( '<STR_LIT>' + self . _local_bind_address + '<STR_LIT::>' <EOL> + str ( self . _local_bind_port ) + '<STR_LIT>' , <EOL> auth = ( self . _cdh_manager_user , self . _cdh_manager_password ) ) . content ) <EOL> result [ '<STR_LIT>' ] = self . extract_cdh_manager_details ( deployments_settings ) [ '<STR_LIT>' ] <EOL> if self . _is_kerberos : <EOL> result [ '<STR_LIT>' ] = result [ '<STR_LIT>' ] <EOL> result [ '<STR_LIT>' ] = self . generate_keytab ( '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = self . generate_keytab ( '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = self . generate_keytab ( '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = self . generate_keytab ( '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = self . generate_base64_for_file ( '<STR_LIT>' , self . _cdh_manager_ip ) <EOL> result [ '<STR_LIT>' ] = self . generate_base64_for_file ( '<STR_LIT>' , self . _cdh_manager_ip ) <EOL> helper = CdhApiHelper ( ApiResource ( self . _local_bind_address , username = self . _cdh_manager_user , password = self . _cdh_manager_password , version = <NUM_LIT:9> ) ) <EOL> hgm_service = helper . get_service_from_cdh ( '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' + helper . get_host ( hgm_service , '<STR_LIT>' ) + '<STR_LIT::>' + helper . get_entry_from_group ( hgm_service , '<STR_LIT>' , '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = helper . get_entry_from_group ( hgm_service , '<STR_LIT>' , '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = helper . get_entry_from_group ( hgm_service , '<STR_LIT>' , '<STR_LIT>' ) <EOL> sentry_service = helper . get_service_from_cdh ( '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = helper . get_entry ( sentry_service , '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = helper . get_host ( sentry_service ) <EOL> result [ '<STR_LIT>' ] = self . generate_keytab ( '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> else : <EOL> result [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> result [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> result [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> result [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> result [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> helper = CdhApiHelper ( ApiResource ( self . _local_bind_address , username = self . _cdh_manager_user , password = self . _cdh_manager_password , version = <NUM_LIT:9> ) ) <EOL> hgm_service = helper . get_service_from_cdh ( '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' + helper . get_host ( hgm_service , '<STR_LIT>' ) + '<STR_LIT::>' + helper . get_entry_from_group ( hgm_service , '<STR_LIT>' , '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = helper . get_entry_from_group ( hgm_service , '<STR_LIT>' , '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = helper . get_entry_from_group ( hgm_service , '<STR_LIT>' , '<STR_LIT>' ) <EOL> master_nodes = self . extract_nodes_info ( '<STR_LIT>' , deployments_settings ) <EOL> for i , node in enumerate ( master_nodes ) : <EOL> result [ '<STR_LIT>' + str ( i + <NUM_LIT:1> ) ] = node [ '<STR_LIT>' ] <EOL> result [ '<STR_LIT>' ] = self . extract_service_namenode ( '<STR_LIT>' , '<STR_LIT>' , deployments_settings ) <EOL> result [ '<STR_LIT>' ] = self . extract_service_namenode ( '<STR_LIT>' , '<STR_LIT>' , deployments_settings ) <EOL> result [ '<STR_LIT>' ] = self . extract_nodes_info ( '<STR_LIT>' , deployments_settings ) [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> result [ '<STR_LIT>' ] = self . extract_nodes_info ( '<STR_LIT>' , deployments_settings ) [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> result [ '<STR_LIT>' ] = self . get_client_config_for_service ( '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = self . get_client_config_for_service ( '<STR_LIT>' ) <EOL> result [ '<STR_LIT>' ] = self . get_client_config_for_service ( '<STR_LIT>' ) <EOL> return result <EOL> def _find_item_by_attr_value ( self , attr_value , attr_name , array_with_dicts ) : <EOL> return next ( item for item in array_with_dicts if item [ attr_name ] == attr_value ) <EOL> def _get_host_ip ( self , host , ansible_ini ) : <EOL> host_info = [ ] <EOL> for line in ansible_ini . split ( '<STR_LIT:\n>' ) : <EOL> if host in line : <EOL> host_info . append ( line . strip ( ) ) <EOL> return host_info [ host_info . index ( '<STR_LIT:[>' + host + '<STR_LIT:]>' ) + <NUM_LIT:1> ] . split ( '<STR_LIT:U+0020>' ) [ <NUM_LIT:1> ] . split ( '<STR_LIT:=>' ) [ <NUM_LIT:1> ] <EOL> def _load_config_yaml ( self , filename ) : <EOL> with open ( filename , '<STR_LIT:r>' ) as stream : <EOL> return yaml . load ( stream ) <EOL> class CdhApiHelper ( object ) : <EOL> def __init__ ( self , cdhApi ) : <EOL> self . cdhApi = cdhApi <EOL> def get_service_from_cdh ( self , name ) : <EOL> cluster = self . cdhApi . get_all_clusters ( ) [ <NUM_LIT:0> ] <EOL> try : <EOL> return next ( service for service in cluster . get_all_services ( ) if service . type == name ) <EOL> except StopIteration : <EOL> raise NoCdhServiceError ( '<STR_LIT>' . format ( name ) ) <EOL> def get_host ( self , service , role = None ) : <EOL> if role is None : <EOL> id = service . get_all_roles ( ) [ <NUM_LIT:0> ] . hostRef . hostId <EOL> else : <EOL> id = service . get_role ( role ) . hostRef . hostId <EOL> return self . cdhApi . get_host ( id ) . hostname <EOL> def get_entry ( self , service , name ) : <EOL> config = service . get_all_roles ( ) [ <NUM_LIT:0> ] . get_config ( '<STR_LIT>' ) <EOL> for config_entry in config : <EOL> if name == config_entry : <EOL> entry = config [ config_entry ] . value or config [ config_entry ] . default <EOL> return entry <EOL> def get_entry_from_group ( self , service , name , group ) : <EOL> config = service . get_role_config_group ( group ) . get_config ( '<STR_LIT>' ) <EOL> for config_entry in config : <EOL> if name == config_entry : <EOL> entry = config [ config_entry ] . value or config [ config_entry ] . default <EOL> return entry <EOL> class NoCdhServiceError ( Exception ) : <EOL> pass </s>
