<s> from __future__ import with_statement <EOL> import os <EOL> import re <EOL> import shutil <EOL> import signal <EOL> import stat <EOL> import subprocess <EOL> import time <EOL> import yaml <EOL> from six import print_ , iteritems <EOL> from ccmlib import common <EOL> from ccmlib . node import Node , NodeError , NodetoolError <EOL> class DseNode ( Node ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , cluster , auto_bootstrap , thrift_interface , storage_interface , jmx_port , remote_debug_port , initial_token , save = True , binary_interface = None , byteman_port = '<STR_LIT:0>' ) : <EOL> super ( DseNode , self ) . __init__ ( name , cluster , auto_bootstrap , thrift_interface , storage_interface , jmx_port , remote_debug_port , initial_token , save , binary_interface , byteman_port ) <EOL> self . get_cassandra_version ( ) <EOL> self . _dse_config_options = { } <EOL> if self . cluster . hasOpscenter ( ) : <EOL> self . _copy_agent ( ) <EOL> def get_install_cassandra_root ( self ) : <EOL> return os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def get_node_cassandra_root ( self ) : <EOL> return os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def get_conf_dir ( self ) : <EOL> """<STR_LIT>""" <EOL> return os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def get_tool ( self , toolname ) : <EOL> return common . join_bin ( os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) , '<STR_LIT>' , toolname ) <EOL> def get_tool_args ( self , toolname ) : <EOL> return [ common . join_bin ( os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) , '<STR_LIT>' , '<STR_LIT>' ) , toolname ] <EOL> def get_env ( self ) : <EOL> ( node_ip , _ ) = self . network_interfaces [ '<STR_LIT>' ] <EOL> return common . make_dse_env ( self . get_install_dir ( ) , self . get_path ( ) , node_ip ) <EOL> def get_cassandra_version ( self ) : <EOL> return common . get_dse_cassandra_version ( self . get_install_dir ( ) ) <EOL> def set_workloads ( self , workloads ) : <EOL> self . workloads = workloads <EOL> self . _update_config ( ) <EOL> if '<STR_LIT>' in self . workloads : <EOL> self . __generate_server_xml ( ) <EOL> if '<STR_LIT>' in self . workloads : <EOL> ( node_ip , _ ) = self . network_interfaces [ '<STR_LIT>' ] <EOL> conf_file = os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> with open ( conf_file , '<STR_LIT:r>' ) as f : <EOL> data = yaml . load ( f ) <EOL> graph_options = data [ '<STR_LIT>' ] <EOL> graph_options [ '<STR_LIT>' ] [ '<STR_LIT:host>' ] = node_ip <EOL> self . set_dse_configuration_options ( { '<STR_LIT>' : graph_options } ) <EOL> self . __update_gremlin_config_yaml ( ) <EOL> if '<STR_LIT>' in self . workloads : <EOL> dsefs_options = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : os . path . join ( self . get_path ( ) , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT:data>' ) } ] } } <EOL> self . set_dse_configuration_options ( dsefs_options ) <EOL> if '<STR_LIT>' in self . workloads : <EOL> self . _update_spark_env ( ) <EOL> def set_dse_configuration_options ( self , values = None ) : <EOL> if values is not None : <EOL> for k , v in iteritems ( values ) : <EOL> self . _dse_config_options [ k ] = v <EOL> self . import_dse_config_files ( ) <EOL> def watch_log_for_alive ( self , nodes , from_mark = None , timeout = <NUM_LIT> , filename = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> super ( DseNode , self ) . watch_log_for_alive ( nodes , from_mark = from_mark , timeout = timeout , filename = filename ) <EOL> def start ( self , <EOL> join_ring = True , <EOL> no_wait = False , <EOL> verbose = False , <EOL> update_pid = True , <EOL> wait_other_notice = False , <EOL> replace_token = None , <EOL> replace_address = None , <EOL> jvm_args = None , <EOL> wait_for_binary_proto = False , <EOL> profile_options = None , <EOL> use_jna = False , <EOL> quiet_start = False , <EOL> allow_root = False ) : <EOL> """<STR_LIT>""" <EOL> if jvm_args is None : <EOL> jvm_args = [ ] <EOL> if self . is_running ( ) : <EOL> raise NodeError ( "<STR_LIT>" % self . name ) <EOL> for itf in list ( self . network_interfaces . values ( ) ) : <EOL> if itf is not None and replace_address is None : <EOL> common . check_socket_available ( itf ) <EOL> if wait_other_notice : <EOL> marks = [ ( node , node . mark_log ( ) ) for node in list ( self . cluster . nodes . values ( ) ) if node . is_running ( ) ] <EOL> self . mark = self . mark_log ( ) <EOL> cdir = self . get_install_dir ( ) <EOL> launch_bin = common . join_bin ( cdir , '<STR_LIT>' , '<STR_LIT>' ) <EOL> shutil . copy ( launch_bin , self . get_bin_dir ( ) ) <EOL> launch_bin = common . join_bin ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if common . is_win ( ) : <EOL> self . __clean_bat ( ) <EOL> if profile_options is not None : <EOL> config = common . get_config ( ) <EOL> if '<STR_LIT>' not in config : <EOL> raise NodeError ( "<STR_LIT>" . format ( common . get_default_path_display_name ( ) ) ) <EOL> cmd = '<STR_LIT>' % config [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in profile_options : <EOL> cmd = cmd + '<STR_LIT:=>' + profile_options [ '<STR_LIT>' ] <EOL> print_ ( cmd ) <EOL> pattern = r'<STR_LIT>' <EOL> common . replace_in_file ( launch_bin , pattern , '<STR_LIT:U+0020>' + pattern + '<STR_LIT:U+0020>' + cmd + '<STR_LIT:">' ) <EOL> os . chmod ( launch_bin , os . stat ( launch_bin ) . st_mode | stat . S_IEXEC ) <EOL> env = self . get_env ( ) <EOL> if common . is_win ( ) : <EOL> self . _clean_win_jmx ( ) <EOL> pidfile = os . path . join ( self . get_path ( ) , '<STR_LIT>' ) <EOL> args = [ launch_bin , '<STR_LIT>' ] <EOL> for workload in self . workloads : <EOL> if '<STR_LIT>' in workload : <EOL> args . append ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in workload : <EOL> args . append ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in workload : <EOL> args . append ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in workload : <EOL> args . append ( '<STR_LIT:-c>' ) <EOL> if '<STR_LIT>' in workload : <EOL> args . append ( '<STR_LIT>' ) <EOL> args += [ '<STR_LIT>' , pidfile , '<STR_LIT>' % str ( join_ring ) ] <EOL> args += [ '<STR_LIT>' % os . path . join ( self . get_path ( ) , '<STR_LIT>' ) ] <EOL> if replace_token is not None : <EOL> args . append ( '<STR_LIT>' % str ( replace_token ) ) <EOL> if replace_address is not None : <EOL> args . append ( '<STR_LIT>' % str ( replace_address ) ) <EOL> if use_jna is False : <EOL> args . append ( '<STR_LIT>' ) <EOL> if allow_root : <EOL> args . append ( '<STR_LIT>' ) <EOL> args = args + jvm_args <EOL> self . _delete_old_pid ( ) <EOL> process = None <EOL> FNULL = open ( os . devnull , '<STR_LIT:w>' ) <EOL> if common . is_win ( ) : <EOL> if ( os . path . isfile ( self . get_path ( ) + "<STR_LIT>" ) ) : <EOL> os . remove ( self . get_path ( ) + "<STR_LIT>" ) <EOL> process = subprocess . Popen ( args , cwd = self . get_bin_dir ( ) , env = env , stdout = FNULL , stderr = subprocess . PIPE ) <EOL> else : <EOL> process = subprocess . Popen ( args , env = env , stdout = FNULL , stderr = subprocess . PIPE ) <EOL> if common . is_win ( ) : <EOL> self . __clean_win_pid ( ) <EOL> self . _update_pid ( process ) <EOL> elif update_pid : <EOL> self . _update_pid ( process ) <EOL> if not self . is_running ( ) : <EOL> raise NodeError ( "<STR_LIT>" % self . name , process ) <EOL> if wait_other_notice : <EOL> for node , mark in marks : <EOL> node . watch_log_for_alive ( self , from_mark = mark ) <EOL> if wait_for_binary_proto : <EOL> self . wait_for_binary_interface ( from_mark = self . mark ) <EOL> if self . cluster . hasOpscenter ( ) : <EOL> self . _start_agent ( ) <EOL> return process <EOL> def stop ( self , wait = True , wait_other_notice = False , gently = True ) : <EOL> stopped = super ( DseNode , self ) . stop ( wait , wait_other_notice , gently ) <EOL> if self . cluster . hasOpscenter ( ) : <EOL> self . _stop_agent ( ) <EOL> return stopped <EOL> def nodetool ( self , cmd , username = None , password = None , capture_output = True , wait = True ) : <EOL> """<STR_LIT>""" <EOL> if capture_output and not wait : <EOL> raise common . ArgumentError ( "<STR_LIT>" ) <EOL> env = self . get_env ( ) <EOL> nodetool = common . join_bin ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> args = [ nodetool , '<STR_LIT>' , '<STR_LIT:localhost>' , '<STR_LIT>' , str ( self . jmx_port ) ] <EOL> if username is not None : <EOL> args += [ '<STR_LIT>' , username ] <EOL> if password is not None : <EOL> args += [ '<STR_LIT>' , password ] <EOL> args += cmd . split ( ) <EOL> if capture_output : <EOL> p = subprocess . Popen ( args , env = env , stdout = subprocess . PIPE , stderr = subprocess . PIPE ) <EOL> stdout , stderr = p . communicate ( ) <EOL> else : <EOL> p = subprocess . Popen ( args , env = env ) <EOL> stdout , stderr = None , None <EOL> if wait : <EOL> exit_status = p . wait ( ) <EOL> if exit_status != <NUM_LIT:0> : <EOL> raise NodetoolError ( "<STR_LIT:U+0020>" . join ( args ) , exit_status , stdout , stderr ) <EOL> return stdout , stderr <EOL> def dsetool ( self , cmd ) : <EOL> env = self . get_env ( ) <EOL> dsetool = common . join_bin ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> args = [ dsetool , '<STR_LIT>' , '<STR_LIT:localhost>' , '<STR_LIT>' , str ( self . jmx_port ) ] <EOL> args += cmd . split ( ) <EOL> p = subprocess . Popen ( args , env = env ) <EOL> p . wait ( ) <EOL> def dse ( self , dse_options = None ) : <EOL> if dse_options is None : <EOL> dse_options = [ ] <EOL> env = self . get_env ( ) <EOL> env [ '<STR_LIT>' ] = self . jmx_port <EOL> dse = common . join_bin ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> args = [ dse ] <EOL> args += dse_options <EOL> p = subprocess . Popen ( args , env = env ) <EOL> p . wait ( ) <EOL> def hadoop ( self , hadoop_options = None ) : <EOL> if hadoop_options is None : <EOL> hadoop_options = [ ] <EOL> env = self . get_env ( ) <EOL> env [ '<STR_LIT>' ] = self . jmx_port <EOL> dse = common . join_bin ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> args = [ dse , '<STR_LIT>' ] <EOL> args += hadoop_options <EOL> p = subprocess . Popen ( args , env = env ) <EOL> p . wait ( ) <EOL> def hive ( self , hive_options = None ) : <EOL> if hive_options is None : <EOL> hive_options = [ ] <EOL> env = self . get_env ( ) <EOL> env [ '<STR_LIT>' ] = self . jmx_port <EOL> dse = common . join_bin ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> args = [ dse , '<STR_LIT>' ] <EOL> args += hive_options <EOL> p = subprocess . Popen ( args , env = env ) <EOL> p . wait ( ) <EOL> def pig ( self , pig_options = None ) : <EOL> if pig_options is None : <EOL> pig_options = [ ] <EOL> env = self . get_env ( ) <EOL> env [ '<STR_LIT>' ] = self . jmx_port <EOL> dse = common . join_bin ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> args = [ dse , '<STR_LIT>' ] <EOL> args += pig_options <EOL> p = subprocess . Popen ( args , env = env ) <EOL> p . wait ( ) <EOL> def sqoop ( self , sqoop_options = None ) : <EOL> if sqoop_options is None : <EOL> sqoop_options = [ ] <EOL> env = self . get_env ( ) <EOL> env [ '<STR_LIT>' ] = self . jmx_port <EOL> dse = common . join_bin ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> args = [ dse , '<STR_LIT>' ] <EOL> args += sqoop_options <EOL> p = subprocess . Popen ( args , env = env ) <EOL> p . wait ( ) <EOL> def spark ( self , spark_options = None ) : <EOL> if spark_options is None : <EOL> spark_options = [ ] <EOL> env = self . get_env ( ) <EOL> env [ '<STR_LIT>' ] = self . jmx_port <EOL> dse = common . join_bin ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> args = [ dse , '<STR_LIT>' ] <EOL> args += spark_options <EOL> p = subprocess . Popen ( args , env = env ) <EOL> p . wait ( ) <EOL> def import_dse_config_files ( self ) : <EOL> self . _update_config ( ) <EOL> if not os . path . isdir ( os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) : <EOL> os . makedirs ( os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> common . copy_directory ( os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . __update_yaml ( ) <EOL> def copy_config_files ( self ) : <EOL> for product in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> src_conf = os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' , product , '<STR_LIT>' ) <EOL> dst_conf = os . path . join ( self . get_path ( ) , '<STR_LIT>' , product , '<STR_LIT>' ) <EOL> if not os . path . isdir ( src_conf ) : <EOL> continue <EOL> if os . path . isdir ( dst_conf ) : <EOL> common . rmdirs ( dst_conf ) <EOL> shutil . copytree ( src_conf , dst_conf ) <EOL> if product == '<STR_LIT>' : <EOL> src_web = os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' , product , '<STR_LIT>' ) <EOL> dst_web = os . path . join ( self . get_path ( ) , '<STR_LIT>' , product , '<STR_LIT>' ) <EOL> if os . path . isdir ( dst_web ) : <EOL> common . rmdirs ( dst_web ) <EOL> shutil . copytree ( src_web , dst_web ) <EOL> if product == '<STR_LIT>' : <EOL> src_lib = os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' , product , '<STR_LIT>' ) <EOL> dst_lib = os . path . join ( self . get_path ( ) , '<STR_LIT>' , product , '<STR_LIT>' ) <EOL> if os . path . isdir ( dst_lib ) : <EOL> common . rmdirs ( dst_lib ) <EOL> shutil . copytree ( src_lib , dst_lib ) <EOL> src_webapps = os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' , product , '<STR_LIT>' ) <EOL> dst_webapps = os . path . join ( self . get_path ( ) , '<STR_LIT>' , product , '<STR_LIT>' ) <EOL> if os . path . isdir ( dst_webapps ) : <EOL> common . rmdirs ( dst_webapps ) <EOL> shutil . copytree ( src_webapps , dst_webapps ) <EOL> src_lib = os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' , product , '<STR_LIT>' , '<STR_LIT>' ) <EOL> dst_lib = os . path . join ( self . get_path ( ) , '<STR_LIT>' , product , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if os . path . isdir ( dst_lib ) : <EOL> common . rmdirs ( dst_lib ) <EOL> if os . path . exists ( src_lib ) : <EOL> shutil . copytree ( src_lib , dst_lib ) <EOL> def import_bin_files ( self ) : <EOL> os . makedirs ( os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> common . copy_directory ( os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' ) , self . get_bin_dir ( ) ) <EOL> common . copy_directory ( os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def _update_log4j ( self ) : <EOL> super ( DseNode , self ) . _update_log4j ( ) <EOL> conf_file = os . path . join ( self . get_conf_dir ( ) , common . LOG4J_CONF ) <EOL> append_pattern = '<STR_LIT>' <EOL> log_file = os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if common . is_win ( ) : <EOL> log_file = re . sub ( "<STR_LIT>" , "<STR_LIT:/>" , log_file ) <EOL> common . replace_in_file ( conf_file , append_pattern , append_pattern + log_file ) <EOL> append_pattern = '<STR_LIT>' <EOL> log_file = os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if common . is_win ( ) : <EOL> log_file = re . sub ( "<STR_LIT>" , "<STR_LIT:/>" , log_file ) <EOL> common . replace_in_file ( conf_file , append_pattern , append_pattern + log_file ) <EOL> append_pattern = '<STR_LIT>' <EOL> log_file = os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if common . is_win ( ) : <EOL> log_file = re . sub ( "<STR_LIT>" , "<STR_LIT:/>" , log_file ) <EOL> common . replace_in_file ( conf_file , append_pattern , append_pattern + log_file ) <EOL> def __update_yaml ( self ) : <EOL> conf_file = os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> with open ( conf_file , '<STR_LIT:r>' ) as f : <EOL> data = yaml . load ( f ) <EOL> data [ '<STR_LIT>' ] = os . path . join ( self . get_path ( ) , '<STR_LIT>' ) <EOL> full_options = common . merge_configuration ( <EOL> self . cluster . _dse_config_options , <EOL> self . _dse_config_options , delete_empty = False ) <EOL> data = common . merge_configuration ( data , full_options ) <EOL> with open ( conf_file , '<STR_LIT:w>' ) as f : <EOL> yaml . safe_dump ( data , f , default_flow_style = False ) <EOL> def _update_log4j ( self ) : <EOL> super ( DseNode , self ) . _update_log4j ( ) <EOL> conf_file = os . path . join ( self . get_conf_dir ( ) , common . LOG4J_CONF ) <EOL> append_pattern = '<STR_LIT>' <EOL> log_file = os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if common . is_win ( ) : <EOL> log_file = re . sub ( "<STR_LIT>" , "<STR_LIT:/>" , log_file ) <EOL> common . replace_in_file ( conf_file , append_pattern , append_pattern + log_file ) <EOL> append_pattern = '<STR_LIT>' <EOL> log_file = os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if common . is_win ( ) : <EOL> log_file = re . sub ( "<STR_LIT>" , "<STR_LIT:/>" , log_file ) <EOL> common . replace_in_file ( conf_file , append_pattern , append_pattern + log_file ) <EOL> append_pattern = '<STR_LIT>' <EOL> log_file = os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if common . is_win ( ) : <EOL> log_file = re . sub ( "<STR_LIT>" , "<STR_LIT:/>" , log_file ) <EOL> common . replace_in_file ( conf_file , append_pattern , append_pattern + log_file ) <EOL> def __generate_server_xml ( self ) : <EOL> server_xml = os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if os . path . isfile ( server_xml ) : <EOL> os . remove ( server_xml ) <EOL> with open ( server_xml , '<STR_LIT>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' % self . network_interfaces [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . close ( ) <EOL> def __update_gremlin_config_yaml ( self ) : <EOL> ( node_ip , _ ) = self . network_interfaces [ '<STR_LIT>' ] <EOL> conf_file = os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> with open ( conf_file , '<STR_LIT:r>' ) as f : <EOL> data = yaml . load ( f ) <EOL> data [ '<STR_LIT>' ] = [ node_ip ] <EOL> with open ( conf_file , '<STR_LIT:w>' ) as f : <EOL> yaml . safe_dump ( data , f , default_flow_style = False ) <EOL> def _get_directories ( self ) : <EOL> dirs = [ ] <EOL> for i in [ '<STR_LIT:data>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , os . path . join ( '<STR_LIT:data>' , '<STR_LIT>' ) ] : <EOL> dirs . append ( os . path . join ( self . get_path ( ) , i ) ) <EOL> return dirs <EOL> def _copy_agent ( self ) : <EOL> agent_source = os . path . join ( self . get_install_dir ( ) , '<STR_LIT>' ) <EOL> agent_target = os . path . join ( self . get_path ( ) , '<STR_LIT>' ) <EOL> if os . path . exists ( agent_source ) and not os . path . exists ( agent_target ) : <EOL> shutil . copytree ( agent_source , agent_target ) <EOL> def _start_agent ( self ) : <EOL> agent_dir = os . path . join ( self . get_path ( ) , '<STR_LIT>' ) <EOL> if os . path . exists ( agent_dir ) : <EOL> self . _write_agent_address_yaml ( agent_dir ) <EOL> self . _write_agent_log4j_properties ( agent_dir ) <EOL> args = [ os . path . join ( agent_dir , '<STR_LIT>' , common . platform_binary ( '<STR_LIT>' ) ) ] <EOL> subprocess . Popen ( args , stdout = subprocess . PIPE , stderr = subprocess . PIPE ) <EOL> def _stop_agent ( self ) : <EOL> agent_dir = os . path . join ( self . get_path ( ) , '<STR_LIT>' ) <EOL> if os . path . exists ( agent_dir ) : <EOL> pidfile = os . path . join ( agent_dir , '<STR_LIT>' ) <EOL> if os . path . exists ( pidfile ) : <EOL> with open ( pidfile , '<STR_LIT:r>' ) as f : <EOL> pid = int ( f . readline ( ) . strip ( ) ) <EOL> f . close ( ) <EOL> if pid is not None : <EOL> try : <EOL> os . kill ( pid , signal . SIGKILL ) <EOL> except OSError : <EOL> pass <EOL> os . remove ( pidfile ) <EOL> def _write_agent_address_yaml ( self , agent_dir ) : <EOL> address_yaml = os . path . join ( agent_dir , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if not os . path . exists ( address_yaml ) : <EOL> with open ( address_yaml , '<STR_LIT>' ) as f : <EOL> ( ip , port ) = self . network_interfaces [ '<STR_LIT>' ] <EOL> jmx = self . jmx_port <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' % ip ) <EOL> f . write ( '<STR_LIT>' % ip ) <EOL> f . write ( '<STR_LIT>' % ip ) <EOL> f . write ( '<STR_LIT>' % os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> f . write ( '<STR_LIT>' % self . get_path ( ) ) <EOL> f . write ( '<STR_LIT>' % os . path . join ( self . get_path ( ) , '<STR_LIT>' ) ) <EOL> f . write ( '<STR_LIT>' % port ) <EOL> f . write ( '<STR_LIT>' % jmx ) <EOL> f . close ( ) <EOL> def _write_agent_log4j_properties ( self , agent_dir ) : <EOL> log4j_properties = os . path . join ( agent_dir , '<STR_LIT>' , '<STR_LIT>' ) <EOL> with open ( log4j_properties , '<STR_LIT>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . close ( ) <EOL> def _update_spark_env ( self ) : <EOL> conf_file = os . path . join ( self . get_path ( ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> env = self . get_env ( ) <EOL> content = [ ] <EOL> with open ( conf_file , '<STR_LIT:r>' ) as f : <EOL> for line in f . readlines ( ) : <EOL> for spark_var in env . keys ( ) : <EOL> if line . startswith ( '<STR_LIT>' % spark_var ) : <EOL> line = '<STR_LIT>' % ( spark_var , env [ spark_var ] ) <EOL> break <EOL> content . append ( line ) <EOL> with open ( conf_file , '<STR_LIT:w>' ) as f : <EOL> f . writelines ( content ) </s>
