<s> import functools <EOL> import uuid <EOL> from oslo_config import cfg <EOL> from oslo_log import log as logging <EOL> import oslo_messaging as messaging <EOL> import six <EOL> from sahara import conductor as c <EOL> from sahara import context <EOL> from sahara import exceptions <EOL> from sahara . i18n import _ <EOL> from sahara . i18n import _LE <EOL> from sahara . plugins import base as plugin_base <EOL> from sahara . service . edp import job_manager <EOL> from sahara . service . health import verification_base as ver_base <EOL> from sahara . service import ntp_service <EOL> from sahara . service import shares <EOL> from sahara . service import trusts <EOL> from sahara . utils import cluster as c_u <EOL> from sahara . utils import remote <EOL> from sahara . utils import rpc as rpc_utils <EOL> conductor = c . API <EOL> CONF = cfg . CONF <EOL> LOG = logging . getLogger ( __name__ ) <EOL> INFRA = None <EOL> def setup_ops ( engine ) : <EOL> global INFRA <EOL> INFRA = engine <EOL> class LocalOps ( object ) : <EOL> def provision_cluster ( self , cluster_id ) : <EOL> context . spawn ( "<STR_LIT>" % cluster_id , <EOL> _provision_cluster , cluster_id ) <EOL> def provision_scaled_cluster ( self , cluster_id , node_group_id_map ) : <EOL> context . spawn ( "<STR_LIT>" % cluster_id , <EOL> _provision_scaled_cluster , cluster_id , node_group_id_map ) <EOL> def terminate_cluster ( self , cluster_id ) : <EOL> context . spawn ( "<STR_LIT>" % cluster_id , <EOL> terminate_cluster , cluster_id ) <EOL> def run_edp_job ( self , job_execution_id ) : <EOL> context . spawn ( "<STR_LIT>" % job_execution_id , <EOL> _run_edp_job , job_execution_id ) <EOL> def cancel_job_execution ( self , job_execution_id ) : <EOL> context . spawn ( "<STR_LIT>" % job_execution_id , <EOL> _cancel_job_execution , job_execution_id ) <EOL> def delete_job_execution ( self , job_execution_id ) : <EOL> context . spawn ( "<STR_LIT>" % job_execution_id , <EOL> _delete_job_execution , job_execution_id ) <EOL> def handle_verification ( self , cluster_id , values ) : <EOL> context . spawn ( '<STR_LIT>' % cluster_id , <EOL> _handle_verification , cluster_id , values ) <EOL> def get_engine_type_and_version ( self ) : <EOL> return INFRA . get_type_and_version ( ) <EOL> def job_execution_suspend ( self , job_execution_id ) : <EOL> context . spawn ( "<STR_LIT>" % job_execution_id , <EOL> _suspend_job_execution , job_execution_id ) <EOL> class RemoteOps ( rpc_utils . RPCClient ) : <EOL> def __init__ ( self ) : <EOL> target = messaging . Target ( topic = '<STR_LIT>' , version = '<STR_LIT:1.0>' ) <EOL> super ( RemoteOps , self ) . __init__ ( target ) <EOL> def provision_cluster ( self , cluster_id ) : <EOL> self . cast ( '<STR_LIT>' , cluster_id = cluster_id ) <EOL> def provision_scaled_cluster ( self , cluster_id , node_group_id_map ) : <EOL> self . cast ( '<STR_LIT>' , cluster_id = cluster_id , <EOL> node_group_id_map = node_group_id_map ) <EOL> def terminate_cluster ( self , cluster_id ) : <EOL> self . cast ( '<STR_LIT>' , cluster_id = cluster_id ) <EOL> def run_edp_job ( self , job_execution_id ) : <EOL> self . cast ( '<STR_LIT>' , job_execution_id = job_execution_id ) <EOL> def cancel_job_execution ( self , job_execution_id ) : <EOL> self . cast ( '<STR_LIT>' , <EOL> job_execution_id = job_execution_id ) <EOL> def delete_job_execution ( self , job_execution_id ) : <EOL> self . cast ( '<STR_LIT>' , <EOL> job_execution_id = job_execution_id ) <EOL> def handle_verification ( self , cluster_id , values ) : <EOL> self . cast ( '<STR_LIT>' , cluster_id = cluster_id , values = values ) <EOL> def get_engine_type_and_version ( self ) : <EOL> return self . call ( '<STR_LIT>' ) <EOL> def job_execution_suspend ( self , job_execution_id ) : <EOL> self . cast ( '<STR_LIT>' , job_execution_id = job_execution_id ) <EOL> def request_context ( func ) : <EOL> @ functools . wraps ( func ) <EOL> def wrapped ( self , ctx , * args , ** kwargs ) : <EOL> context . set_ctx ( context . Context ( ** ctx ) ) <EOL> return func ( self , * args , ** kwargs ) <EOL> return wrapped <EOL> class OpsServer ( rpc_utils . RPCServer ) : <EOL> def __init__ ( self ) : <EOL> target = messaging . Target ( topic = '<STR_LIT>' , server = uuid . uuid4 ( ) , <EOL> version = '<STR_LIT:1.0>' ) <EOL> super ( OpsServer , self ) . __init__ ( target ) <EOL> @ request_context <EOL> def provision_cluster ( self , cluster_id ) : <EOL> _provision_cluster ( cluster_id ) <EOL> @ request_context <EOL> def provision_scaled_cluster ( self , cluster_id , node_group_id_map ) : <EOL> _provision_scaled_cluster ( cluster_id , node_group_id_map ) <EOL> @ request_context <EOL> def terminate_cluster ( self , cluster_id ) : <EOL> terminate_cluster ( cluster_id ) <EOL> @ request_context <EOL> def run_edp_job ( self , job_execution_id ) : <EOL> _run_edp_job ( job_execution_id ) <EOL> @ request_context <EOL> def cancel_job_execution ( self , job_execution_id ) : <EOL> _cancel_job_execution ( job_execution_id ) <EOL> @ request_context <EOL> def delete_job_execution ( self , job_execution_id ) : <EOL> _delete_job_execution ( job_execution_id ) <EOL> @ request_context <EOL> def handle_verification ( self , cluster_id , values ) : <EOL> _handle_verification ( cluster_id , values ) <EOL> @ request_context <EOL> def get_engine_type_and_version ( self ) : <EOL> return INFRA . get_type_and_version ( ) <EOL> @ request_context <EOL> def job_execution_suspend ( self , job_execution_id ) : <EOL> _suspend_job_execution ( job_execution_id ) <EOL> def _setup_trust_for_cluster ( cluster ) : <EOL> cluster = conductor . cluster_get ( context . ctx ( ) , cluster ) <EOL> trusts . create_trust_for_cluster ( cluster ) <EOL> trusts . use_os_admin_auth_token ( cluster ) <EOL> def ops_error_handler ( description ) : <EOL> def decorator ( f ) : <EOL> @ functools . wraps ( f ) <EOL> def wrapper ( cluster_id , * args , ** kwds ) : <EOL> ctx = context . ctx ( ) <EOL> try : <EOL> c_u . change_cluster_status_description ( cluster_id , "<STR_LIT>" ) <EOL> f ( cluster_id , * args , ** kwds ) <EOL> except Exception as ex : <EOL> cluster = conductor . cluster_get ( ctx , cluster_id ) <EOL> if ( cluster is None or <EOL> cluster . status == c_u . CLUSTER_STATUS_DELETING ) : <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return <EOL> msg = six . text_type ( ex ) <EOL> LOG . exception ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) . format ( reason = msg ) ) <EOL> try : <EOL> desc = description . format ( reason = msg ) <EOL> if _rollback_cluster ( cluster , ex ) : <EOL> c_u . change_cluster_status ( <EOL> cluster , c_u . CLUSTER_STATUS_ACTIVE , desc ) <EOL> else : <EOL> c_u . change_cluster_status ( <EOL> cluster , c_u . CLUSTER_STATUS_ERROR , desc ) <EOL> except Exception as rex : <EOL> cluster = conductor . cluster_get ( ctx , cluster_id ) <EOL> if ( cluster is None or <EOL> cluster . status == c_u . CLUSTER_STATUS_DELETING ) : <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return <EOL> LOG . exception ( <EOL> _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) . format ( reason = six . text_type ( rex ) ) ) <EOL> desc = "<STR_LIT>" . format ( msg , six . text_type ( rex ) ) <EOL> c_u . change_cluster_status ( <EOL> cluster , c_u . CLUSTER_STATUS_ERROR , <EOL> description . format ( reason = desc ) ) <EOL> return wrapper <EOL> return decorator <EOL> def _rollback_cluster ( cluster , reason ) : <EOL> _setup_trust_for_cluster ( cluster ) <EOL> context . set_step_type ( _ ( "<STR_LIT>" ) ) <EOL> return INFRA . rollback_cluster ( cluster , reason ) <EOL> def _prepare_provisioning ( cluster_id ) : <EOL> ctx = context . ctx ( ) <EOL> cluster = conductor . cluster_get ( ctx , cluster_id ) <EOL> plugin = plugin_base . PLUGINS . get_plugin ( cluster . plugin_name ) <EOL> for nodegroup in cluster . node_groups : <EOL> update_dict = { } <EOL> update_dict [ "<STR_LIT>" ] = INFRA . get_node_group_image_username ( <EOL> nodegroup ) <EOL> conductor . node_group_update ( ctx , nodegroup , update_dict ) <EOL> _setup_trust_for_cluster ( cluster ) <EOL> cluster = conductor . cluster_get ( ctx , cluster_id ) <EOL> return ctx , cluster , plugin <EOL> def _update_sahara_info ( ctx , cluster ) : <EOL> sahara_info = { <EOL> '<STR_LIT>' : INFRA . get_type_and_version ( ) , <EOL> '<STR_LIT>' : remote . get_remote_type_and_version ( ) } <EOL> return conductor . cluster_update ( <EOL> ctx , cluster , { '<STR_LIT>' : sahara_info } ) <EOL> @ ops_error_handler ( <EOL> _ ( "<STR_LIT>" ) ) <EOL> def _provision_cluster ( cluster_id ) : <EOL> ctx , cluster , plugin = _prepare_provisioning ( cluster_id ) <EOL> cluster = _update_sahara_info ( ctx , cluster ) <EOL> cluster = c_u . change_cluster_status ( <EOL> cluster , c_u . CLUSTER_STATUS_INFRAUPDATING ) <EOL> plugin . update_infra ( cluster ) <EOL> cluster = conductor . cluster_get ( ctx , cluster_id ) <EOL> context . set_step_type ( _ ( "<STR_LIT>" ) ) <EOL> INFRA . create_cluster ( cluster ) <EOL> cluster = c_u . change_cluster_status ( <EOL> cluster , c_u . CLUSTER_STATUS_CONFIGURING ) <EOL> context . set_step_type ( _ ( "<STR_LIT>" ) ) <EOL> if hasattr ( plugin , '<STR_LIT>' ) : <EOL> plugin . validate_images ( cluster , reconcile = True ) <EOL> shares . mount_shares ( cluster ) <EOL> plugin . configure_cluster ( cluster ) <EOL> ntp_service . configure_ntp ( cluster_id ) <EOL> cluster = c_u . change_cluster_status ( <EOL> cluster , c_u . CLUSTER_STATUS_STARTING ) <EOL> context . set_step_type ( _ ( "<STR_LIT>" ) ) <EOL> plugin . start_cluster ( cluster ) <EOL> cluster = c_u . change_cluster_status ( <EOL> cluster , c_u . CLUSTER_STATUS_ACTIVE ) <EOL> for je in conductor . job_execution_get_all ( ctx , cluster_id = cluster . id ) : <EOL> job_manager . run_job ( je . id ) <EOL> _refresh_health_for_cluster ( cluster_id ) <EOL> @ ops_error_handler ( <EOL> _ ( "<STR_LIT>" ) ) <EOL> def _provision_scaled_cluster ( cluster_id , node_group_id_map ) : <EOL> ctx , cluster , plugin = _prepare_provisioning ( cluster_id ) <EOL> cluster = c_u . change_cluster_status ( <EOL> cluster , c_u . CLUSTER_STATUS_DECOMMISSIONING ) <EOL> instances_to_delete = [ ] <EOL> for node_group in cluster . node_groups : <EOL> new_count = node_group_id_map [ node_group . id ] <EOL> if new_count < node_group . count : <EOL> instances_to_delete += node_group . instances [ new_count : <EOL> node_group . count ] <EOL> if instances_to_delete : <EOL> context . set_step_type ( _ ( "<STR_LIT>" ) ) <EOL> plugin . decommission_nodes ( cluster , instances_to_delete ) <EOL> cluster = c_u . change_cluster_status ( <EOL> cluster , c_u . CLUSTER_STATUS_SCALING ) <EOL> context . set_step_type ( _ ( "<STR_LIT>" ) ) <EOL> instance_ids = INFRA . scale_cluster ( cluster , node_group_id_map ) <EOL> if instance_ids : <EOL> ntp_service . configure_ntp ( cluster_id ) <EOL> cluster = c_u . change_cluster_status ( <EOL> cluster , c_u . CLUSTER_STATUS_CONFIGURING ) <EOL> instances = c_u . get_instances ( cluster , instance_ids ) <EOL> context . set_step_type ( _ ( "<STR_LIT>" ) ) <EOL> plugin . scale_cluster ( cluster , instances ) <EOL> c_u . change_cluster_status ( cluster , c_u . CLUSTER_STATUS_ACTIVE ) <EOL> _refresh_health_for_cluster ( cluster_id ) <EOL> @ ops_error_handler ( <EOL> _ ( "<STR_LIT>" ) ) <EOL> def terminate_cluster ( cluster_id ) : <EOL> ctx = context . ctx ( ) <EOL> _setup_trust_for_cluster ( cluster_id ) <EOL> job_manager . update_job_statuses ( cluster_id = cluster_id ) <EOL> cluster = conductor . cluster_get ( ctx , cluster_id ) <EOL> plugin = plugin_base . PLUGINS . get_plugin ( cluster . plugin_name ) <EOL> context . set_step_type ( _ ( "<STR_LIT>" ) ) <EOL> plugin . on_terminate_cluster ( cluster ) <EOL> context . set_step_type ( _ ( "<STR_LIT>" ) ) <EOL> INFRA . shutdown_cluster ( cluster ) <EOL> trusts . delete_trust_from_cluster ( cluster ) <EOL> conductor . cluster_destroy ( ctx , cluster ) <EOL> def _run_edp_job ( job_execution_id ) : <EOL> job_manager . run_job ( job_execution_id ) <EOL> def _suspend_job_execution ( job_execution_id ) : <EOL> job_manager . suspend_job ( job_execution_id ) <EOL> def _cancel_job_execution ( job_execution_id ) : <EOL> job_manager . cancel_job ( job_execution_id ) <EOL> def _delete_job_execution ( job_execution_id ) : <EOL> try : <EOL> job_execution = job_manager . cancel_job ( job_execution_id ) <EOL> if not job_execution : <EOL> return <EOL> except exceptions . CancelingFailed : <EOL> LOG . error ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> conductor . job_execution_destroy ( context . ctx ( ) , job_execution_id ) <EOL> def _refresh_health_for_cluster ( cluster_id ) : <EOL> st_dict = { '<STR_LIT>' : { '<STR_LIT:status>' : '<STR_LIT>' } } <EOL> try : <EOL> ver_base . validate_verification_start ( cluster_id ) <EOL> ver_base . handle_verification ( cluster_id , st_dict ) <EOL> except ver_base . CannotVerifyError : <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> except Exception : <EOL> LOG . debug ( "<STR_LIT>" ) <EOL> ver_base . clean_verification_data ( cluster_id ) <EOL> def _handle_verification ( cluster_id , values ) : <EOL> ver_base . handle_verification ( cluster_id , values ) </s>
