<s> import os <EOL> import time <EOL> import errno <EOL> import json <EOL> import subprocess <EOL> from os import path <EOL> import requests <EOL> import pika <EOL> from cloudify import ctx <EOL> from cloudify . decorators import operation <EOL> from cloudify . exceptions import NonRecoverableError , HttpException <EOL> from cloudify . manager import get_resource_from_manager <EOL> from cloudify import broker_config , utils <EOL> from riemann_controller import config <EOL> RIEMANN_CONFIGS_DIR = '<STR_LIT>' <EOL> RIEMANN_LOG_PATH = '<STR_LIT>' <EOL> @ operation <EOL> def create ( policy_types = None , <EOL> policy_triggers = None , <EOL> groups = None , <EOL> ** _ ) : <EOL> policy_types = policy_types or { } <EOL> groups = groups or { } <EOL> policy_triggers = policy_triggers or { } <EOL> _process_types_and_triggers ( groups , policy_types , policy_triggers ) <EOL> deployment_config_dir_path = _deployment_config_dir ( ) <EOL> if not os . path . isdir ( deployment_config_dir_path ) : <EOL> os . makedirs ( deployment_config_dir_path ) <EOL> os . chmod ( deployment_config_dir_path , <NUM_LIT:0> <NUM_LIT> ) <EOL> with open ( _deployment_config_template ( ) ) as f : <EOL> deployment_config_template = f . read ( ) <EOL> with open ( os . path . join ( _deployment_config_dir ( ) , '<STR_LIT>' ) , '<STR_LIT:w>' ) as f : <EOL> f . write ( json . dumps ( groups ) ) <EOL> with open ( path . join ( deployment_config_dir_path , <EOL> '<STR_LIT>' ) , '<STR_LIT:w>' ) as f : <EOL> f . write ( config . create ( ctx , <EOL> policy_types , <EOL> policy_triggers , <EOL> groups , <EOL> deployment_config_template ) ) <EOL> _publish_configuration_event ( '<STR_LIT:start>' , deployment_config_dir_path ) <EOL> _verify_core_up ( deployment_config_dir_path ) <EOL> @ operation <EOL> def delete ( ** _ ) : <EOL> deployment_config_dir_path = _deployment_config_dir ( ) <EOL> _publish_configuration_event ( '<STR_LIT>' , deployment_config_dir_path ) <EOL> def _deployment_config_dir ( ) : <EOL> return os . path . join ( os . environ [ RIEMANN_CONFIGS_DIR ] , <EOL> ctx . deployment . id ) <EOL> def _publish_configuration_event ( state , deployment_config_dir_path ) : <EOL> manager_queue = '<STR_LIT>' <EOL> exchange_name = '<STR_LIT>' <EOL> broker_host = broker_config . broker_hostname <EOL> broker_port , ssl_options = utils . internal . get_broker_ssl_and_port ( <EOL> ssl_enabled = broker_config . broker_ssl_enabled , <EOL> cert_path = broker_config . broker_cert_path , <EOL> ) <EOL> credentials = pika . credentials . PlainCredentials ( <EOL> username = broker_config . broker_username , <EOL> password = broker_config . broker_password , <EOL> ) <EOL> connection = pika . BlockingConnection ( <EOL> pika . ConnectionParameters ( host = broker_host , <EOL> port = broker_port , <EOL> credentials = credentials , <EOL> ssl = broker_config . broker_ssl_enabled , <EOL> ssl_options = ssl_options ) <EOL> ) <EOL> try : <EOL> channel = connection . channel ( ) <EOL> channel . exchange_declare ( exchange = exchange_name , <EOL> type = '<STR_LIT>' , <EOL> durable = False , <EOL> auto_delete = True , <EOL> internal = False ) <EOL> channel . queue_declare ( <EOL> queue = manager_queue , <EOL> auto_delete = True , <EOL> durable = False , <EOL> exclusive = False ) <EOL> channel . queue_bind ( exchange = exchange_name , <EOL> queue = manager_queue , <EOL> routing_key = manager_queue ) <EOL> channel . basic_publish ( <EOL> exchange = exchange_name , <EOL> routing_key = manager_queue , <EOL> body = json . dumps ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:state>' : state , <EOL> '<STR_LIT>' : deployment_config_dir_path , <EOL> '<STR_LIT>' : ctx . deployment . id , <EOL> '<STR_LIT:time>' : int ( time . time ( ) ) <EOL> } ) ) <EOL> finally : <EOL> connection . close ( ) <EOL> def _deployment_config_template ( ) : <EOL> return path . abspath ( path . join ( path . dirname ( __file__ ) , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) ) <EOL> def _verify_core_up ( deployment_config_dir_path ) : <EOL> timeout = ctx . bootstrap_context . policy_engine . start_timeout or <NUM_LIT:30> <EOL> ok_path = path . join ( deployment_config_dir_path , '<STR_LIT>' ) <EOL> end = time . time ( ) + timeout <EOL> while time . time ( ) < end : <EOL> try : <EOL> with ( open ( ok_path ) ) : <EOL> pass <EOL> return <EOL> except IOError , e : <EOL> if e . errno in [ errno . ENOENT ] : <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> else : <EOL> raise <EOL> try : <EOL> riemann_log_output = subprocess . check_output ( <EOL> '<STR_LIT>' . format ( RIEMANN_LOG_PATH ) , shell = True ) <EOL> except Exception as e : <EOL> riemann_log_output = '<STR_LIT>' . format ( e ) <EOL> raise NonRecoverableError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( timeout , <EOL> RIEMANN_LOG_PATH , <EOL> riemann_log_output ) ) <EOL> def _process_types_and_triggers ( groups , policy_types , policy_triggers ) : <EOL> types_to_process = set ( ) <EOL> types_to_remove = set ( ) <EOL> triggers_to_process = set ( ) <EOL> triggers_to_remove = set ( ) <EOL> for group in groups . values ( ) : <EOL> for policy in group [ '<STR_LIT>' ] . values ( ) : <EOL> types_to_process . add ( policy [ '<STR_LIT:type>' ] ) <EOL> for trigger in policy [ '<STR_LIT>' ] . values ( ) : <EOL> triggers_to_process . add ( trigger [ '<STR_LIT:type>' ] ) <EOL> for policy_type_name , policy_type in policy_types . items ( ) : <EOL> if policy_type_name in types_to_process : <EOL> policy_type [ '<STR_LIT:source>' ] = _process_source ( policy_type [ '<STR_LIT:source>' ] ) <EOL> else : <EOL> types_to_remove . add ( policy_type_name ) <EOL> for policy_trigger_name , trigger in policy_triggers . items ( ) : <EOL> if policy_trigger_name in triggers_to_process : <EOL> trigger [ '<STR_LIT:source>' ] = _process_source ( trigger [ '<STR_LIT:source>' ] ) <EOL> else : <EOL> triggers_to_remove . add ( policy_trigger_name ) <EOL> for policy_type_name in types_to_remove : <EOL> del policy_types [ policy_type_name ] <EOL> for trigger_type_name in triggers_to_remove : <EOL> del policy_triggers [ trigger_type_name ] <EOL> def _process_source ( source ) : <EOL> split = source . split ( '<STR_LIT>' ) <EOL> schema = split [ <NUM_LIT:0> ] <EOL> the_rest = '<STR_LIT>' . join ( split [ <NUM_LIT:1> : ] ) <EOL> try : <EOL> if schema in [ '<STR_LIT:http>' , '<STR_LIT>' ] : <EOL> return requests . get ( source ) . text <EOL> elif schema == '<STR_LIT:file>' and the_rest : <EOL> with open ( the_rest ) as f : <EOL> return f . read ( ) <EOL> except IOError , e : <EOL> raise NonRecoverableError ( '<STR_LIT>' <EOL> . format ( source , e . message ) ) <EOL> try : <EOL> return ctx . get_resource ( source ) <EOL> except HttpException : <EOL> pass <EOL> try : <EOL> return get_resource_from_manager ( source ) <EOL> except HttpException : <EOL> pass <EOL> raise NonRecoverableError ( '<STR_LIT>' <EOL> . format ( source ) ) </s>
