<s> import logging <EOL> import requests <EOL> import six <EOL> import time <EOL> import urllib3 <EOL> from oslo_serialization import jsonutils <EOL> from nailgun import consts <EOL> from nailgun . db import db <EOL> from nailgun . db . sqlalchemy import models <EOL> from nailgun import objects <EOL> from nailgun . settings import settings <EOL> from nailgun . statistics . fuel_statistics . installation_info import InstallationInfo <EOL> from nailgun . statistics . utils import dithered <EOL> from nailgun . statistics . utils import prepare_logger <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> class StatsSender ( object ) : <EOL> def build_collector_url ( self , url_template ) : <EOL> return getattr ( settings , url_template ) . format ( collector_server = settings . COLLECTOR_SERVER ) <EOL> def ping_collector ( self ) : <EOL> try : <EOL> resp = requests . get ( self . build_collector_url ( "<STR_LIT>" ) , <EOL> timeout = settings . COLLECTOR_RESP_TIMEOUT ) <EOL> resp . raise_for_status ( ) <EOL> return True <EOL> except ( urllib3 . exceptions . DecodeError , <EOL> urllib3 . exceptions . ProxyError , <EOL> requests . exceptions . ConnectionError , <EOL> requests . exceptions . Timeout , <EOL> requests . exceptions . TooManyRedirects , <EOL> requests . exceptions . HTTPError ) as e : <EOL> logger . error ( "<STR_LIT>" , type ( e ) . __name__ ) <EOL> except Exception as e : <EOL> logger . exception ( "<STR_LIT>" , six . text_type ( e ) ) <EOL> return False <EOL> def send_data_to_url ( self , url , data ) : <EOL> resp = None <EOL> try : <EOL> headers = { <EOL> '<STR_LIT>' : '<STR_LIT:application/json>' , <EOL> '<STR_LIT>' : InstallationInfo ( ) . get_master_node_uid ( ) <EOL> } <EOL> resp = requests . post ( <EOL> url , <EOL> headers = headers , <EOL> data = jsonutils . dumps ( data ) , <EOL> timeout = settings . COLLECTOR_RESP_TIMEOUT ) <EOL> except ( urllib3 . exceptions . DecodeError , <EOL> urllib3 . exceptions . ProxyError , <EOL> requests . exceptions . ConnectionError , <EOL> requests . exceptions . Timeout , <EOL> requests . exceptions . TooManyRedirects ) as e : <EOL> logger . error ( "<STR_LIT>" , <EOL> type ( e ) . __name__ ) <EOL> except Exception as e : <EOL> logger . exception ( <EOL> "<STR_LIT>" , six . text_type ( e ) ) <EOL> return resp <EOL> def is_status_acceptable ( <EOL> self , resp_status_code , resp_status , <EOL> codes = ( requests . codes . created , requests . codes . ok ) <EOL> ) : <EOL> return ( resp_status_code in codes and <EOL> resp_status == consts . LOG_CHUNK_SEND_STATUS . ok ) <EOL> def send_log_serialized ( self , records , ids ) : <EOL> if records : <EOL> logger . info ( "<STR_LIT>" , len ( records ) ) <EOL> resp = self . send_data_to_url ( <EOL> url = self . build_collector_url ( "<STR_LIT>" ) , <EOL> data = { "<STR_LIT>" : records } <EOL> ) <EOL> resp_dict = resp . json ( ) <EOL> if self . is_status_acceptable ( resp . status_code , <EOL> resp_dict [ "<STR_LIT:status>" ] ) : <EOL> records_resp = resp_dict [ "<STR_LIT>" ] <EOL> saved_ids = set ( ) <EOL> failed_ids = set ( ) <EOL> skipped_ids = set ( ) <EOL> for record in records_resp : <EOL> if record [ "<STR_LIT:status>" ] == consts . LOG_RECORD_SEND_STATUS . failed : <EOL> failed_ids . add ( record [ "<STR_LIT>" ] ) <EOL> elif record [ "<STR_LIT:status>" ] == consts . LOG_RECORD_SEND_STATUS . skipped : <EOL> skipped_ids . add ( record [ "<STR_LIT>" ] ) <EOL> else : <EOL> saved_ids . add ( record [ "<STR_LIT>" ] ) <EOL> sent_saved_ids = set ( saved_ids ) & set ( ids ) <EOL> logger . info ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> six . text_type ( list ( sent_saved_ids ) ) , <EOL> six . text_type ( list ( failed_ids ) ) , <EOL> six . text_type ( list ( skipped_ids ) ) ) <EOL> if sent_saved_ids : <EOL> db ( ) . query ( models . ActionLog ) . filter ( <EOL> models . ActionLog . id . in_ ( sent_saved_ids ) <EOL> ) . update ( <EOL> { "<STR_LIT>" : True } , synchronize_session = False <EOL> ) <EOL> db ( ) . commit ( ) <EOL> else : <EOL> logger . error ( "<STR_LIT>" , <EOL> six . text_type ( resp . text ) ) <EOL> def send_action_log ( self ) : <EOL> action_log = db ( ) . query ( models . ActionLog ) . order_by ( <EOL> models . ActionLog . id <EOL> ) . filter_by ( <EOL> is_sent = False <EOL> ) . limit ( settings . STATS_SEND_COUNT ) <EOL> logger . info ( "<STR_LIT>" , action_log . count ( ) ) <EOL> uid = InstallationInfo ( ) . get_master_node_uid ( ) <EOL> offset = <NUM_LIT:0> <EOL> while True : <EOL> log_chunk = action_log . offset ( offset ) <EOL> records = [ ] <EOL> ids = [ ] <EOL> logger . info ( "<STR_LIT>" , six . text_type ( log_chunk . count ( ) ) ) <EOL> for log_record in log_chunk : <EOL> body = objects . ActionLog . to_dict ( log_record ) <EOL> record = { <EOL> '<STR_LIT>' : body [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : uid , <EOL> '<STR_LIT:body>' : body <EOL> } <EOL> records . append ( record ) <EOL> ids . append ( log_record . id ) <EOL> self . send_log_serialized ( records , ids ) <EOL> if log_chunk . count ( ) < settings . STATS_SEND_COUNT : <EOL> break <EOL> offset += settings . STATS_SEND_COUNT <EOL> def send_installation_info ( self ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> inst_info = InstallationInfo ( ) . get_installation_info ( ) <EOL> resp = self . send_data_to_url ( <EOL> url = self . build_collector_url ( "<STR_LIT>" ) , <EOL> data = { "<STR_LIT>" : inst_info } <EOL> ) <EOL> resp_dict = resp . json ( ) <EOL> if self . is_status_acceptable ( resp . status_code , resp_dict [ "<STR_LIT:status>" ] ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> else : <EOL> logger . error ( "<STR_LIT>" , <EOL> six . text_type ( resp . text ) ) <EOL> def send_oswl_serialized ( self , rec_data , ids ) : <EOL> if rec_data : <EOL> resp = self . send_data_to_url ( <EOL> url = self . build_collector_url ( "<STR_LIT>" ) , <EOL> data = { "<STR_LIT>" : rec_data } <EOL> ) <EOL> resp_dict = resp . json ( ) <EOL> if self . is_status_acceptable ( resp . status_code , <EOL> resp_dict [ "<STR_LIT:status>" ] ) : <EOL> records_resp = resp_dict [ "<STR_LIT>" ] <EOL> status_failed = consts . LOG_RECORD_SEND_STATUS . failed <EOL> saved_ids = set ( r [ "<STR_LIT:id>" ] for r in records_resp <EOL> if r [ "<STR_LIT:status>" ] != status_failed ) <EOL> failed_ids = set ( r [ "<STR_LIT:id>" ] for r in records_resp <EOL> if r [ "<STR_LIT:status>" ] == status_failed ) <EOL> sent_saved_ids = set ( saved_ids ) & set ( ids ) <EOL> logger . info ( "<STR_LIT>" , <EOL> six . text_type ( list ( sent_saved_ids ) ) , <EOL> six . text_type ( list ( failed_ids ) ) ) <EOL> if sent_saved_ids : <EOL> db ( ) . query ( models . OpenStackWorkloadStats ) . filter ( <EOL> models . OpenStackWorkloadStats . id . in_ ( sent_saved_ids ) <EOL> ) . update ( <EOL> { "<STR_LIT>" : True } , synchronize_session = False <EOL> ) <EOL> db ( ) . commit ( ) <EOL> else : <EOL> logger . error ( "<STR_LIT>" , <EOL> six . text_type ( resp . text ) ) <EOL> def send_oswl_info ( self ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> oswl_data = objects . OpenStackWorkloadStatsCollection . get_ready_to_send ( ) . limit ( settings . OSWL_SEND_COUNT ) <EOL> logger . info ( "<STR_LIT>" , <EOL> six . text_type ( oswl_data . count ( ) ) ) <EOL> uid = InstallationInfo ( ) . get_master_node_uid ( ) <EOL> offset = <NUM_LIT:0> <EOL> records = [ ] <EOL> ids = [ ] <EOL> while True : <EOL> data_chunk = oswl_data . offset ( offset ) <EOL> for rec in data_chunk : <EOL> rec_data = objects . OpenStackWorkloadStats . to_dict ( rec ) <EOL> rec_data [ '<STR_LIT>' ] = uid <EOL> rec_data [ '<STR_LIT>' ] = rec_data [ '<STR_LIT>' ] . isoformat ( ) <EOL> rec_data [ '<STR_LIT>' ] = rec_data [ '<STR_LIT>' ] . isoformat ( ) <EOL> records . append ( rec_data ) <EOL> ids . append ( rec_data [ '<STR_LIT:id>' ] ) <EOL> self . send_oswl_serialized ( records , ids ) <EOL> if data_chunk . count ( ) < settings . OSWL_SEND_COUNT : <EOL> break <EOL> offset += settings . OSWL_SEND_COUNT <EOL> logger . info ( "<STR_LIT>" ) <EOL> def send_stats_once ( self ) : <EOL> try : <EOL> if objects . MasterNodeSettings . must_send_stats ( ) : <EOL> if self . ping_collector ( ) : <EOL> self . send_action_log ( ) <EOL> self . send_installation_info ( ) <EOL> self . send_oswl_info ( ) <EOL> time . sleep ( dithered ( settings . STATS_SEND_INTERVAL ) ) <EOL> else : <EOL> time . sleep ( dithered ( settings . COLLECTOR_PING_INTERVAL ) ) <EOL> else : <EOL> time . sleep ( dithered ( settings . STATS_ENABLE_CHECK_INTERVAL ) ) <EOL> except Exception as e : <EOL> logger . error ( "<STR_LIT>" , six . text_type ( e ) ) <EOL> time . sleep ( dithered ( settings . COLLECTOR_PING_INTERVAL ) ) <EOL> finally : <EOL> db . remove ( ) <EOL> def run ( ) : <EOL> prepare_logger ( logger , file_name = '<STR_LIT>' ) <EOL> logger . info ( "<STR_LIT>" ) <EOL> try : <EOL> while True : <EOL> StatsSender ( ) . send_stats_once ( ) <EOL> except ( KeyboardInterrupt , SystemExit ) : <EOL> logger . info ( "<STR_LIT>" ) </s>
