<s> from __future__ import print_function <EOL> import contextlib <EOL> import copy <EOL> import datetime <EOL> import errno <EOL> import fcntl <EOL> import glob <EOL> import hashlib <EOL> import importlib <EOL> import io <EOL> import json <EOL> import logging <EOL> import math <EOL> import os <EOL> import pwd <EOL> import re <EOL> import shlex <EOL> import signal <EOL> import sys <EOL> import tempfile <EOL> import threading <EOL> from functools import wraps <EOL> from subprocess import PIPE <EOL> from subprocess import Popen <EOL> from subprocess import STDOUT <EOL> import dateutil . tz <EOL> import docker <EOL> import service_configuration_lib <EOL> import yaml <EOL> from docker import Client <EOL> from docker . utils import kwargs_from_env <EOL> from kazoo . client import KazooClient <EOL> SPACER = '<STR_LIT:.>' <EOL> INFRA_ZK_PATH = '<STR_LIT>' <EOL> PATH_TO_SYSTEM_PAASTA_CONFIG_DIR = os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> DEFAULT_SOA_DIR = service_configuration_lib . DEFAULT_SOA_DIR <EOL> DEFAULT_DOCKERCFG_LOCATION = "<STR_LIT>" <EOL> DEPLOY_PIPELINE_NON_DEPLOY_STEPS = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ) <EOL> ANY_CLUSTER = '<STR_LIT>' <EOL> ANY_INSTANCE = '<STR_LIT>' <EOL> DEFAULT_LOGLEVEL = '<STR_LIT>' <EOL> no_escape = re . compile ( '<STR_LIT>' ) <EOL> DEFAULT_SYNAPSE_HAPROXY_URL_FORMAT = "<STR_LIT>" <EOL> log = logging . getLogger ( '<STR_LIT:__main__>' ) <EOL> class InvalidInstanceConfig ( Exception ) : <EOL> pass <EOL> class InstanceConfig ( dict ) : <EOL> def __init__ ( self , cluster , instance , service , config_dict , branch_dict ) : <EOL> self . config_dict = config_dict <EOL> self . branch_dict = branch_dict <EOL> self . cluster = cluster <EOL> self . instance = instance <EOL> self . service = service <EOL> config_interpolation_keys = ( '<STR_LIT>' , ) <EOL> interpolation_facts = self . __get_interpolation_facts ( ) <EOL> for key in config_interpolation_keys : <EOL> if key in self . config_dict : <EOL> self . config_dict [ key ] = self . config_dict [ key ] . format ( ** interpolation_facts ) <EOL> def __get_interpolation_facts ( self ) : <EOL> return { <EOL> '<STR_LIT>' : self . cluster , <EOL> '<STR_LIT>' : self . instance , <EOL> '<STR_LIT>' : self . service , <EOL> } <EOL> def get_cluster ( self ) : <EOL> return self . cluster <EOL> def get_instance ( self ) : <EOL> return self . instance <EOL> def get_service ( self ) : <EOL> return self . service <EOL> def get_branch ( self ) : <EOL> return SPACER . join ( ( self . get_cluster ( ) , self . get_instance ( ) ) ) <EOL> def get_deploy_group ( self ) : <EOL> return self . config_dict . get ( '<STR_LIT>' , self . get_branch ( ) ) <EOL> def get_mem ( self ) : <EOL> """<STR_LIT>""" <EOL> mem = self . config_dict . get ( '<STR_LIT>' , <NUM_LIT> ) <EOL> return mem <EOL> def get_mem_swap ( self ) : <EOL> """<STR_LIT>""" <EOL> mem = self . get_mem ( ) <EOL> mem_swap = int ( math . ceil ( mem ) ) <EOL> return "<STR_LIT>" % mem_swap <EOL> def get_cpus ( self ) : <EOL> """<STR_LIT>""" <EOL> cpus = self . config_dict . get ( '<STR_LIT>' , <NUM_LIT> ) <EOL> return cpus <EOL> def get_disk ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = self . config_dict . get ( '<STR_LIT>' , <NUM_LIT> ) <EOL> return disk <EOL> def get_cmd ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config_dict . get ( '<STR_LIT>' , None ) <EOL> def get_env_dictionary ( self ) : <EOL> """<STR_LIT>""" <EOL> env = { <EOL> "<STR_LIT>" : self . service , <EOL> "<STR_LIT>" : self . instance , <EOL> "<STR_LIT>" : self . cluster , <EOL> } <EOL> user_env = self . config_dict . get ( '<STR_LIT>' , { } ) <EOL> env . update ( user_env ) <EOL> return env <EOL> def get_env ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . get_env_dictionary ( ) <EOL> def get_args ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . get_cmd ( ) is None : <EOL> return self . config_dict . get ( '<STR_LIT:args>' , [ ] ) <EOL> else : <EOL> args = self . config_dict . get ( '<STR_LIT:args>' , None ) <EOL> if args is None : <EOL> return args <EOL> else : <EOL> raise InvalidInstanceConfig ( '<STR_LIT>' ) <EOL> def get_monitoring ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config_dict . get ( '<STR_LIT>' , { } ) <EOL> def get_deploy_blacklist ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config_dict . get ( '<STR_LIT>' , [ ] ) <EOL> def get_deploy_whitelist ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config_dict . get ( '<STR_LIT>' , [ ] ) <EOL> def get_monitoring_blacklist ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config_dict . get ( '<STR_LIT>' , self . get_deploy_blacklist ( ) ) <EOL> def get_docker_image ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . branch_dict . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def get_desired_state ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . branch_dict . get ( '<STR_LIT>' , '<STR_LIT:start>' ) <EOL> def get_force_bounce ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . branch_dict . get ( '<STR_LIT>' , None ) <EOL> def check_cpus ( self ) : <EOL> cpus = self . get_cpus ( ) <EOL> if cpus is not None : <EOL> if not isinstance ( cpus , ( float , int ) ) : <EOL> return False , '<STR_LIT>' % cpus <EOL> return True , '<STR_LIT>' <EOL> def check_mem ( self ) : <EOL> mem = self . get_mem ( ) <EOL> if mem is not None : <EOL> if not isinstance ( mem , ( float , int ) ) : <EOL> return False , '<STR_LIT>' % mem <EOL> return True , '<STR_LIT>' <EOL> def check_disk ( self ) : <EOL> disk = self . get_disk ( ) <EOL> if disk is not None : <EOL> if not isinstance ( disk , ( float , int ) ) : <EOL> return False , '<STR_LIT>' % disk <EOL> return True , '<STR_LIT>' <EOL> def check ( self , param ) : <EOL> check_methods = { <EOL> '<STR_LIT>' : self . check_cpus , <EOL> '<STR_LIT>' : self . check_mem , <EOL> } <EOL> if param in check_methods : <EOL> return check_methods [ param ] ( ) <EOL> else : <EOL> return False , '<STR_LIT>' % param <EOL> def validate ( self ) : <EOL> error_msgs = [ ] <EOL> for param in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> check_passed , check_msg = self . check ( param ) <EOL> if not check_passed : <EOL> error_msgs . append ( check_msg ) <EOL> return error_msgs <EOL> def get_extra_volumes ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config_dict . get ( '<STR_LIT>' , [ ] ) <EOL> def get_pool ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config_dict . get ( '<STR_LIT>' , '<STR_LIT:default>' ) <EOL> def get_net ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config_dict . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def validate_service_instance ( service , instance , cluster , soa_dir ) : <EOL> marathon_services = get_services_for_cluster ( cluster = cluster , instance_type = '<STR_LIT>' , soa_dir = soa_dir ) <EOL> chronos_services = get_services_for_cluster ( cluster = cluster , instance_type = '<STR_LIT>' , soa_dir = soa_dir ) <EOL> if ( service , instance ) in marathon_services : <EOL> return '<STR_LIT>' <EOL> elif ( service , instance ) in chronos_services : <EOL> return '<STR_LIT>' <EOL> else : <EOL> print ( "<STR_LIT>" <EOL> % ( compose_job_id ( service , instance ) , cluster ) ) <EOL> sys . exit ( <NUM_LIT:3> ) <EOL> class PaastaColors : <EOL> """<STR_LIT>""" <EOL> BLUE = '<STR_LIT>' <EOL> BOLD = '<STR_LIT>' <EOL> CYAN = '<STR_LIT>' <EOL> DEFAULT = '<STR_LIT>' <EOL> GREEN = '<STR_LIT>' <EOL> GREY = '<STR_LIT>' <EOL> MAGENTA = '<STR_LIT>' <EOL> RED = '<STR_LIT>' <EOL> YELLOW = '<STR_LIT>' <EOL> @ staticmethod <EOL> def bold ( text ) : <EOL> """<STR_LIT>""" <EOL> return PaastaColors . color_text ( PaastaColors . BOLD , text ) <EOL> @ staticmethod <EOL> def blue ( text ) : <EOL> """<STR_LIT>""" <EOL> return PaastaColors . color_text ( PaastaColors . BLUE , text ) <EOL> @ staticmethod <EOL> def green ( text ) : <EOL> """<STR_LIT>""" <EOL> return PaastaColors . color_text ( PaastaColors . GREEN , text ) <EOL> @ staticmethod <EOL> def red ( text ) : <EOL> """<STR_LIT>""" <EOL> return PaastaColors . color_text ( PaastaColors . RED , text ) <EOL> @ staticmethod <EOL> def magenta ( text ) : <EOL> """<STR_LIT>""" <EOL> return PaastaColors . color_text ( PaastaColors . MAGENTA , text ) <EOL> @ staticmethod <EOL> def color_text ( color , text ) : <EOL> """<STR_LIT>""" <EOL> replaced = text . replace ( PaastaColors . DEFAULT , PaastaColors . DEFAULT + color ) <EOL> return color + replaced + PaastaColors . DEFAULT <EOL> @ staticmethod <EOL> def cyan ( text ) : <EOL> """<STR_LIT>""" <EOL> return PaastaColors . color_text ( PaastaColors . CYAN , text ) <EOL> @ staticmethod <EOL> def yellow ( text ) : <EOL> """<STR_LIT>""" <EOL> return PaastaColors . color_text ( PaastaColors . YELLOW , text ) <EOL> @ staticmethod <EOL> def grey ( text ) : <EOL> return PaastaColors . color_text ( PaastaColors . GREY , text ) <EOL> @ staticmethod <EOL> def default ( text ) : <EOL> return PaastaColors . color_text ( PaastaColors . DEFAULT , text ) <EOL> LOG_COMPONENTS = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : PaastaColors . blue , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : PaastaColors . cyan , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : PaastaColors . green , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : PaastaColors . magenta , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : PaastaColors . magenta , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> } <EOL> class NoSuchLogComponent ( Exception ) : <EOL> pass <EOL> def validate_log_component ( component ) : <EOL> if component in LOG_COMPONENTS . keys ( ) : <EOL> return True <EOL> else : <EOL> raise NoSuchLogComponent <EOL> def get_git_url ( service , soa_dir = DEFAULT_SOA_DIR ) : <EOL> """<STR_LIT>""" <EOL> general_config = service_configuration_lib . read_service_configuration ( <EOL> service , <EOL> soa_dir = soa_dir , <EOL> ) <EOL> default_location = '<STR_LIT>' % service <EOL> return general_config . get ( '<STR_LIT>' , default_location ) <EOL> class NoSuchLogLevel ( Exception ) : <EOL> pass <EOL> _log_writer = None <EOL> _log_writer_classes = { } <EOL> def register_log_writer ( name ) : <EOL> """<STR_LIT>""" <EOL> def outer ( bounce_func ) : <EOL> _log_writer_classes [ name ] = bounce_func <EOL> return bounce_func <EOL> return outer <EOL> def get_log_writer_class ( name ) : <EOL> return _log_writer_classes [ name ] <EOL> def list_log_writers ( ) : <EOL> return _log_writer_classes . keys ( ) <EOL> def configure_log ( ) : <EOL> """<STR_LIT>""" <EOL> log_writer_options = load_system_paasta_config ( ) . get_log_writer ( ) <EOL> global _log_writer <EOL> LogWriterClass = get_log_writer_class ( log_writer_options [ '<STR_LIT>' ] ) <EOL> _log_writer = LogWriterClass ( ** log_writer_options ) <EOL> def _log ( * args , ** kwargs ) : <EOL> if _log_writer is None : <EOL> configure_log ( ) <EOL> return _log_writer . log ( * args , ** kwargs ) <EOL> class LogWriter ( object ) : <EOL> def log ( self , line , component , level = DEFAULT_LOGLEVEL , cluster = ANY_CLUSTER , instance = ANY_INSTANCE ) : <EOL> raise NotImplementedError ( ) <EOL> def _now ( ) : <EOL> return datetime . datetime . utcnow ( ) . isoformat ( ) <EOL> def remove_ansi_escape_sequences ( line ) : <EOL> """<STR_LIT>""" <EOL> return no_escape . sub ( '<STR_LIT>' , line ) <EOL> def format_log_line ( level , cluster , service , instance , component , line , timestamp = None ) : <EOL> """<STR_LIT>""" <EOL> validate_log_component ( component ) <EOL> if not timestamp : <EOL> timestamp = _now ( ) <EOL> line = remove_ansi_escape_sequences ( line ) <EOL> message = json . dumps ( { <EOL> '<STR_LIT>' : timestamp , <EOL> '<STR_LIT>' : level , <EOL> '<STR_LIT>' : cluster , <EOL> '<STR_LIT>' : service , <EOL> '<STR_LIT>' : instance , <EOL> '<STR_LIT>' : component , <EOL> '<STR_LIT:message>' : line , <EOL> } , sort_keys = True ) <EOL> return message <EOL> def get_log_name_for_service ( service ) : <EOL> return '<STR_LIT>' % service <EOL> @ register_log_writer ( '<STR_LIT>' ) <EOL> class ScribeLogWriter ( LogWriter ) : <EOL> def __init__ ( self , scribe_host = '<STR_LIT>' , scribe_port = <NUM_LIT> , scribe_disable = False , ** kwargs ) : <EOL> self . clog = importlib . import_module ( '<STR_LIT>' ) <EOL> self . clog . config . configure ( scribe_host = scribe_host , scribe_port = scribe_port , scribe_disable = scribe_disable ) <EOL> def log ( self , service , line , component , level = DEFAULT_LOGLEVEL , cluster = ANY_CLUSTER , instance = ANY_INSTANCE ) : <EOL> """<STR_LIT>""" <EOL> if level == '<STR_LIT>' : <EOL> print ( line , file = sys . stdout ) <EOL> elif level == '<STR_LIT>' : <EOL> print ( line , file = sys . stderr ) <EOL> else : <EOL> raise NoSuchLogLevel <EOL> log_name = get_log_name_for_service ( service ) <EOL> formatted_line = format_log_line ( level , cluster , service , instance , component , line ) <EOL> self . clog . log_line ( log_name , formatted_line ) <EOL> @ register_log_writer ( '<STR_LIT:null>' ) <EOL> class NullLogWriter ( LogWriter ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ** kwargs ) : <EOL> pass <EOL> def log ( self , service , line , component , level = DEFAULT_LOGLEVEL , cluster = ANY_CLUSTER , instance = ANY_INSTANCE ) : <EOL> pass <EOL> @ register_log_writer ( '<STR_LIT:file>' ) <EOL> class FileLogWriter ( LogWriter ) : <EOL> def __init__ ( self , path_format , mode = '<STR_LIT>' , line_delimeter = '<STR_LIT:\n>' , flock = False ) : <EOL> self . path_format = path_format <EOL> self . mode = mode <EOL> self . flock = flock <EOL> self . line_delimeter = line_delimeter <EOL> @ contextlib . contextmanager <EOL> def maybe_flock ( self , fd ) : <EOL> if self . flock : <EOL> try : <EOL> fcntl . flock ( fd , fcntl . LOCK_EX ) <EOL> yield <EOL> finally : <EOL> fcntl . flock ( fd , fcntl . LOCK_UN ) <EOL> else : <EOL> yield <EOL> def format_path ( self , service , component , level , cluster , instance ) : <EOL> return self . path_format . format ( <EOL> service = service , <EOL> component = component , <EOL> level = level , <EOL> cluster = cluster , <EOL> instance = instance , <EOL> ) <EOL> def log ( self , service , line , component , level = DEFAULT_LOGLEVEL , cluster = ANY_CLUSTER , instance = ANY_INSTANCE ) : <EOL> path = self . format_path ( service , component , level , cluster , instance ) <EOL> to_write = "<STR_LIT>" % ( format_log_line ( level , cluster , service , instance , component , line ) , self . line_delimeter ) <EOL> with io . FileIO ( path , mode = self . mode , closefd = True ) as f : <EOL> with self . maybe_flock ( f ) : <EOL> f . write ( to_write ) <EOL> def _timeout ( process ) : <EOL> """<STR_LIT>""" <EOL> if process . poll ( ) is None : <EOL> try : <EOL> process . kill ( ) <EOL> except OSError as e : <EOL> if e . errno != errno . ESRCH : <EOL> raise <EOL> class PaastaNotConfiguredError ( Exception ) : <EOL> pass <EOL> class NoConfigurationForServiceError ( Exception ) : <EOL> pass <EOL> def get_readable_files_in_glob ( input_glob ) : <EOL> """<STR_LIT>""" <EOL> files = [ ] <EOL> for f in sorted ( glob . glob ( input_glob ) ) : <EOL> if os . path . isfile ( f ) and os . access ( f , os . R_OK ) : <EOL> files . append ( f ) <EOL> return files <EOL> def load_system_paasta_config ( path = PATH_TO_SYSTEM_PAASTA_CONFIG_DIR ) : <EOL> """<STR_LIT>""" <EOL> config = { } <EOL> if not os . path . isdir ( path ) : <EOL> raise PaastaNotConfiguredError ( "<STR_LIT>" % path ) <EOL> if not os . access ( path , os . R_OK ) : <EOL> raise PaastaNotConfiguredError ( "<STR_LIT>" % path ) <EOL> try : <EOL> for config_file in get_readable_files_in_glob ( "<STR_LIT>" % path ) : <EOL> with open ( os . path . join ( path , config_file ) ) as f : <EOL> config . update ( json . load ( f ) ) <EOL> except IOError as e : <EOL> raise PaastaNotConfiguredError ( "<STR_LIT>" % ( e . filename , e . strerror ) ) <EOL> return SystemPaastaConfig ( config , path ) <EOL> class SystemPaastaConfig ( dict ) : <EOL> def __init__ ( self , config , directory ) : <EOL> self . directory = directory <EOL> super ( SystemPaastaConfig , self ) . __init__ ( config ) <EOL> def get_zk_hosts ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> hosts = self [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise PaastaNotConfiguredError ( '<STR_LIT>' <EOL> % self . directory ) <EOL> if hosts . startswith ( '<STR_LIT>' ) : <EOL> return hosts [ len ( '<STR_LIT>' ) : ] <EOL> return hosts <EOL> def get_docker_registry ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise PaastaNotConfiguredError ( '<STR_LIT>' <EOL> % self . directory ) <EOL> def get_volumes ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise PaastaNotConfiguredError ( '<STR_LIT>' % self . directory ) <EOL> def get_cluster ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise PaastaNotConfiguredError ( '<STR_LIT>' % self . directory ) <EOL> def get_scribe_map ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise PaastaNotConfiguredError ( '<STR_LIT>' % self . directory ) <EOL> def get_fsm_cluster_map ( self ) : <EOL> try : <EOL> return self [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise PaastaNotConfiguredError ( <EOL> '<STR_LIT>' % self . directory ) <EOL> def get_dashboard_links ( self ) : <EOL> try : <EOL> return self [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise PaastaNotConfiguredError ( <EOL> '<STR_LIT>' % self . directory ) <EOL> def get_fsm_deploy_pipeline ( self ) : <EOL> try : <EOL> return self [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise PaastaNotConfiguredError ( <EOL> '<STR_LIT>' % self . directory ) <EOL> def get_log_writer ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise PaastaNotConfiguredError ( '<STR_LIT>' % self . directory ) <EOL> def get_log_reader ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise PaastaNotConfiguredError ( '<STR_LIT>' % self . directory ) <EOL> def get_sensu_host ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . get ( '<STR_LIT>' , '<STR_LIT:localhost>' ) <EOL> def get_sensu_port ( self ) : <EOL> """<STR_LIT>""" <EOL> return int ( self . get ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> def get_dockercfg_location ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . get ( '<STR_LIT>' , DEFAULT_DOCKERCFG_LOCATION ) <EOL> def get_synapse_port ( self ) : <EOL> """<STR_LIT>""" <EOL> return int ( self . get ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> def get_default_synapse_host ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . get ( '<STR_LIT>' , '<STR_LIT:localhost>' ) <EOL> def get_synapse_haproxy_url_format ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . get ( '<STR_LIT>' , DEFAULT_SYNAPSE_HAPROXY_URL_FORMAT ) <EOL> def get_cluster_autoscaling_resources ( self ) : <EOL> return self . get ( '<STR_LIT>' , { } ) <EOL> def _run ( command , env = os . environ , timeout = None , log = False , stream = False , stdin = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> output = [ ] <EOL> if log : <EOL> service = kwargs [ '<STR_LIT>' ] <EOL> component = kwargs [ '<STR_LIT>' ] <EOL> cluster = kwargs . get ( '<STR_LIT>' , ANY_CLUSTER ) <EOL> instance = kwargs . get ( '<STR_LIT>' , ANY_INSTANCE ) <EOL> loglevel = kwargs . get ( '<STR_LIT>' , DEFAULT_LOGLEVEL ) <EOL> try : <EOL> process = Popen ( shlex . split ( command ) , stdout = PIPE , stderr = STDOUT , stdin = stdin , env = env ) <EOL> process . name = command <EOL> if timeout : <EOL> proctimer = threading . Timer ( timeout , _timeout , ( process , ) ) <EOL> proctimer . start ( ) <EOL> for line in iter ( process . stdout . readline , '<STR_LIT>' ) : <EOL> if stream : <EOL> print ( line . rstrip ( '<STR_LIT:\n>' ) ) <EOL> if log : <EOL> _log ( <EOL> service = service , <EOL> line = line . rstrip ( '<STR_LIT:\n>' ) , <EOL> component = component , <EOL> level = loglevel , <EOL> cluster = cluster , <EOL> instance = instance , <EOL> ) <EOL> output . append ( line . rstrip ( '<STR_LIT:\n>' ) ) <EOL> returncode = process . wait ( ) <EOL> except OSError as e : <EOL> if log : <EOL> _log ( <EOL> service = service , <EOL> line = e . strerror . rstrip ( '<STR_LIT:\n>' ) , <EOL> component = component , <EOL> level = loglevel , <EOL> cluster = cluster , <EOL> instance = instance , <EOL> ) <EOL> output . append ( e . strerror . rstrip ( '<STR_LIT:\n>' ) ) <EOL> returncode = e . errno <EOL> except ( KeyboardInterrupt , SystemExit ) : <EOL> if timeout : <EOL> proctimer . cancel ( ) <EOL> raise <EOL> else : <EOL> if timeout : <EOL> proctimer . cancel ( ) <EOL> if returncode == - <NUM_LIT:9> : <EOL> output . append ( "<STR_LIT>" % ( command , timeout ) ) <EOL> return returncode , '<STR_LIT:\n>' . join ( output ) <EOL> def get_umask ( ) : <EOL> """<STR_LIT>""" <EOL> old_umask = os . umask ( <NUM_LIT> <NUM_LIT> ) <EOL> os . umask ( old_umask ) <EOL> return old_umask <EOL> @ contextlib . contextmanager <EOL> def atomic_file_write ( target_path ) : <EOL> dirname = os . path . dirname ( target_path ) <EOL> basename = os . path . basename ( target_path ) <EOL> with tempfile . NamedTemporaryFile ( <EOL> dir = dirname , <EOL> prefix = ( '<STR_LIT>' % basename ) , <EOL> delete = False <EOL> ) as f : <EOL> temp_target_path = f . name <EOL> yield f <EOL> mode = <NUM_LIT:0> <NUM_LIT> & ( ~ get_umask ( ) ) <EOL> os . chmod ( temp_target_path , mode ) <EOL> os . rename ( temp_target_path , target_path ) <EOL> class InvalidJobNameError ( Exception ) : <EOL> pass <EOL> def compose_job_id ( name , instance , git_hash = None , config_hash = None , spacer = SPACER ) : <EOL> """<STR_LIT>""" <EOL> composed = '<STR_LIT>' % ( name , spacer , instance ) <EOL> if git_hash and config_hash : <EOL> composed = '<STR_LIT>' % ( composed , spacer , git_hash , spacer , config_hash ) <EOL> elif git_hash or config_hash : <EOL> raise InvalidJobNameError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( git_hash , config_hash ) ) <EOL> return composed <EOL> def decompose_job_id ( job_id , spacer = SPACER ) : <EOL> """<STR_LIT>""" <EOL> decomposed = job_id . split ( spacer ) <EOL> if len ( decomposed ) == <NUM_LIT:2> : <EOL> git_hash = None <EOL> config_hash = None <EOL> elif len ( decomposed ) == <NUM_LIT:4> : <EOL> git_hash = decomposed [ <NUM_LIT:2> ] <EOL> config_hash = decomposed [ <NUM_LIT:3> ] <EOL> else : <EOL> raise InvalidJobNameError ( '<STR_LIT>' % job_id ) <EOL> return ( decomposed [ <NUM_LIT:0> ] , decomposed [ <NUM_LIT:1> ] , git_hash , config_hash ) <EOL> def build_docker_image_name ( upstream_job_name ) : <EOL> """<STR_LIT>""" <EOL> docker_registry_url = load_system_paasta_config ( ) . get_docker_registry ( ) <EOL> name = '<STR_LIT>' % ( docker_registry_url , upstream_job_name ) <EOL> return name <EOL> def build_docker_tag ( upstream_job_name , upstream_git_commit ) : <EOL> """<STR_LIT>""" <EOL> tag = '<STR_LIT>' % ( <EOL> build_docker_image_name ( upstream_job_name ) , <EOL> upstream_git_commit , <EOL> ) <EOL> return tag <EOL> def check_docker_image ( service , tag ) : <EOL> """<STR_LIT>""" <EOL> docker_client = docker . Client ( timeout = <NUM_LIT> ) <EOL> image_name = build_docker_image_name ( service ) <EOL> docker_tag = build_docker_tag ( service , tag ) <EOL> images = docker_client . images ( name = image_name ) <EOL> result = [ image for image in images if docker_tag in image [ '<STR_LIT>' ] ] <EOL> if len ( result ) > <NUM_LIT:1> : <EOL> raise ValueError ( '<STR_LIT>' % docker_tag , result ) <EOL> return len ( result ) == <NUM_LIT:1> <EOL> def datetime_from_utc_to_local ( utc_datetime ) : <EOL> return datetime_convert_timezone ( utc_datetime , dateutil . tz . tzutc ( ) , dateutil . tz . tzlocal ( ) ) <EOL> def datetime_convert_timezone ( datetime , from_zone , to_zone ) : <EOL> datetime = datetime . replace ( tzinfo = from_zone ) <EOL> converted_datetime = datetime . astimezone ( to_zone ) <EOL> converted_datetime = converted_datetime . replace ( tzinfo = None ) <EOL> return converted_datetime <EOL> def get_username ( ) : <EOL> """<STR_LIT>""" <EOL> return os . environ . get ( '<STR_LIT>' , pwd . getpwuid ( os . getuid ( ) ) [ <NUM_LIT:0> ] ) <EOL> def get_default_cluster_for_service ( service ) : <EOL> cluster = None <EOL> try : <EOL> cluster = load_system_paasta_config ( ) . get_cluster ( ) <EOL> except PaastaNotConfiguredError : <EOL> clusters_deployed_to = list_clusters ( service ) <EOL> if len ( clusters_deployed_to ) > <NUM_LIT:0> : <EOL> cluster = clusters_deployed_to [ <NUM_LIT:0> ] <EOL> else : <EOL> raise NoConfigurationForServiceError ( "<STR_LIT>" % service ) <EOL> return cluster <EOL> def get_soa_cluster_deploy_files ( service = None , soa_dir = DEFAULT_SOA_DIR , instance_type = None ) : <EOL> if service is None : <EOL> service = '<STR_LIT:*>' <EOL> service_path = os . path . join ( soa_dir , service ) <EOL> if instance_type == '<STR_LIT>' or instance_type == '<STR_LIT>' : <EOL> instance_types = instance_type <EOL> else : <EOL> instance_types = '<STR_LIT>' <EOL> search_re = r'<STR_LIT>' + instance_types + r'<STR_LIT>' <EOL> for yaml_file in glob . glob ( '<STR_LIT>' % service_path ) : <EOL> cluster_re_match = re . search ( search_re , yaml_file ) <EOL> if cluster_re_match is not None : <EOL> cluster = cluster_re_match . group ( <NUM_LIT:2> ) <EOL> yield ( cluster , yaml_file ) <EOL> def list_clusters ( service = None , soa_dir = DEFAULT_SOA_DIR , instance_type = None ) : <EOL> """<STR_LIT>""" <EOL> clusters = set ( ) <EOL> for cluster , _ in get_soa_cluster_deploy_files ( <EOL> service = service , <EOL> soa_dir = soa_dir , <EOL> instance_type = instance_type , <EOL> ) : <EOL> clusters . add ( cluster ) <EOL> return sorted ( clusters ) <EOL> def list_all_instances_for_service ( service , clusters = None , instance_type = None , soa_dir = DEFAULT_SOA_DIR ) : <EOL> instances = set ( ) <EOL> if not clusters : <EOL> clusters = list_clusters ( service , soa_dir = soa_dir ) <EOL> for cluster in clusters : <EOL> for service_instance in get_service_instance_list ( service , cluster , instance_type , soa_dir = soa_dir ) : <EOL> instances . add ( service_instance [ <NUM_LIT:1> ] ) <EOL> return instances <EOL> def get_service_instance_list ( service , cluster = None , instance_type = None , soa_dir = DEFAULT_SOA_DIR ) : <EOL> """<STR_LIT>""" <EOL> if not cluster : <EOL> cluster = load_system_paasta_config ( ) . get_cluster ( ) <EOL> if instance_type == '<STR_LIT>' or instance_type == '<STR_LIT>' : <EOL> instance_types = [ instance_type ] <EOL> else : <EOL> instance_types = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> instance_list = [ ] <EOL> for srv_instance_type in instance_types : <EOL> conf_file = "<STR_LIT>" % ( srv_instance_type , cluster ) <EOL> log . info ( "<STR_LIT>" % ( soa_dir , conf_file ) ) <EOL> instances = service_configuration_lib . read_extra_service_information ( <EOL> service , <EOL> conf_file , <EOL> soa_dir = soa_dir <EOL> ) <EOL> for instance in instances : <EOL> instance_list . append ( ( service , instance ) ) <EOL> log . debug ( "<STR_LIT>" , instance_list ) <EOL> return instance_list <EOL> def get_services_for_cluster ( cluster = None , instance_type = None , soa_dir = DEFAULT_SOA_DIR ) : <EOL> """<STR_LIT>""" <EOL> if not cluster : <EOL> cluster = load_system_paasta_config ( ) . get_cluster ( ) <EOL> rootdir = os . path . abspath ( soa_dir ) <EOL> log . info ( "<STR_LIT>" , rootdir , cluster ) <EOL> instance_list = [ ] <EOL> for srv_dir in os . listdir ( rootdir ) : <EOL> instance_list . extend ( get_service_instance_list ( srv_dir , cluster , instance_type , soa_dir ) ) <EOL> return instance_list <EOL> def parse_yaml_file ( yaml_file ) : <EOL> return yaml . load ( open ( yaml_file ) ) <EOL> def get_docker_host ( ) : <EOL> return os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def get_docker_client ( ) : <EOL> client_opts = kwargs_from_env ( assert_hostname = False ) <EOL> if '<STR_LIT>' in client_opts : <EOL> return Client ( ** client_opts ) <EOL> else : <EOL> return Client ( base_url = get_docker_host ( ) , ** client_opts ) <EOL> def get_running_mesos_docker_containers ( ) : <EOL> client = get_docker_client ( ) <EOL> running_containers = client . containers ( ) <EOL> return [ container for container in running_containers if "<STR_LIT>" in container [ "<STR_LIT>" ] [ <NUM_LIT:0> ] ] <EOL> class TimeoutError ( Exception ) : <EOL> pass <EOL> def timeout ( seconds = <NUM_LIT:10> , error_message = os . strerror ( errno . ETIME ) ) : <EOL> def decorator ( func ) : <EOL> def _handle_timeout ( signum , frame ) : <EOL> raise TimeoutError ( error_message ) <EOL> def wrapper ( * args , ** kwargs ) : <EOL> signal . signal ( signal . SIGALRM , _handle_timeout ) <EOL> signal . alarm ( seconds ) <EOL> try : <EOL> result = func ( * args , ** kwargs ) <EOL> finally : <EOL> signal . alarm ( <NUM_LIT:0> ) <EOL> return result <EOL> return wraps ( func ) ( wrapper ) <EOL> return decorator <EOL> class Timeout : <EOL> def __init__ ( self , seconds = <NUM_LIT:1> , error_message = '<STR_LIT>' ) : <EOL> self . seconds = seconds <EOL> self . error_message = error_message <EOL> def handle_timeout ( self , signum , frame ) : <EOL> raise TimeoutError ( self . error_message ) <EOL> def __enter__ ( self ) : <EOL> signal . signal ( signal . SIGALRM , self . handle_timeout ) <EOL> signal . alarm ( self . seconds ) <EOL> def __exit__ ( self , type , value , traceback ) : <EOL> signal . alarm ( <NUM_LIT:0> ) <EOL> def print_with_indent ( line , indent = <NUM_LIT:2> ) : <EOL> """<STR_LIT>""" <EOL> print ( "<STR_LIT:U+0020>" * indent + line ) <EOL> class NoDeploymentsAvailable ( Exception ) : <EOL> pass <EOL> def load_deployments_json ( service , soa_dir = DEFAULT_SOA_DIR ) : <EOL> deployment_file = os . path . join ( soa_dir , service , '<STR_LIT>' ) <EOL> if os . path . isfile ( deployment_file ) : <EOL> with open ( deployment_file ) as f : <EOL> return DeploymentsJson ( json . load ( f ) [ '<STR_LIT>' ] ) <EOL> else : <EOL> raise NoDeploymentsAvailable <EOL> class DeploymentsJson ( dict ) : <EOL> def get_branch_dict ( self , service , branch ) : <EOL> full_branch = '<STR_LIT>' % ( service , branch ) <EOL> return self . get ( full_branch , { } ) <EOL> def get_paasta_branch_from_deploy_group ( identifier ) : <EOL> return '<STR_LIT>' % ( identifier ) <EOL> def get_paasta_branch ( cluster , instance ) : <EOL> return get_paasta_branch_from_deploy_group ( '<STR_LIT>' % ( cluster , instance ) ) <EOL> def parse_timestamp ( tstamp ) : <EOL> return datetime . datetime . strptime ( tstamp , '<STR_LIT>' ) <EOL> def format_timestamp ( dt = None ) : <EOL> if dt is None : <EOL> dt = datetime . datetime . utcnow ( ) <EOL> return dt . strftime ( '<STR_LIT>' ) <EOL> def get_paasta_tag_from_deploy_group ( identifier , desired_state ) : <EOL> timestamp = format_timestamp ( datetime . datetime . utcnow ( ) ) <EOL> return '<STR_LIT>' % ( identifier , timestamp , desired_state ) <EOL> def get_paasta_tag ( cluster , instance , desired_state ) : <EOL> timestamp = format_timestamp ( datetime . datetime . utcnow ( ) ) <EOL> return '<STR_LIT>' % ( cluster , instance , timestamp , desired_state ) <EOL> def format_tag ( tag ) : <EOL> return '<STR_LIT>' % tag <EOL> class NoDockerImageError ( Exception ) : <EOL> pass <EOL> def get_docker_url ( registry_uri , docker_image ) : <EOL> """<STR_LIT>""" <EOL> if not docker_image : <EOL> raise NoDockerImageError ( '<STR_LIT>' ) <EOL> docker_url = '<STR_LIT>' % ( registry_uri , docker_image ) <EOL> return docker_url <EOL> def get_config_hash ( config , force_bounce = None ) : <EOL> """<STR_LIT>""" <EOL> hasher = hashlib . md5 ( ) <EOL> hasher . update ( json . dumps ( config , sort_keys = True ) + ( force_bounce or '<STR_LIT>' ) ) <EOL> return "<STR_LIT>" % hasher . hexdigest ( ) [ : <NUM_LIT:8> ] <EOL> def get_code_sha_from_dockerurl ( docker_url ) : <EOL> """<STR_LIT>""" <EOL> parts = docker_url . split ( '<STR_LIT:->' ) <EOL> return "<STR_LIT>" % parts [ - <NUM_LIT:1> ] [ : <NUM_LIT:8> ] <EOL> def is_under_replicated ( num_available , expected_count , crit_threshold ) : <EOL> """<STR_LIT>""" <EOL> if expected_count == <NUM_LIT:0> : <EOL> ratio = <NUM_LIT:100> <EOL> else : <EOL> ratio = ( num_available / float ( expected_count ) ) * <NUM_LIT:100> <EOL> if ratio < crit_threshold : <EOL> return ( True , ratio ) <EOL> else : <EOL> return ( False , ratio ) <EOL> def deploy_blacklist_to_constraints ( deploy_blacklist ) : <EOL> """<STR_LIT>""" <EOL> constraints = [ ] <EOL> for blacklisted_location in deploy_blacklist : <EOL> constraints . append ( [ blacklisted_location [ <NUM_LIT:0> ] , "<STR_LIT>" , blacklisted_location [ <NUM_LIT:1> ] ] ) <EOL> return constraints <EOL> def deploy_whitelist_to_constraints ( deploy_whitelist ) : <EOL> """<STR_LIT>""" <EOL> if len ( deploy_whitelist ) > <NUM_LIT:0> : <EOL> ( region_type , regions ) = deploy_whitelist <EOL> regionstr = '<STR_LIT:|>' . join ( regions ) <EOL> return [ [ region_type , '<STR_LIT>' , regionstr ] ] <EOL> return [ ] <EOL> def terminal_len ( text ) : <EOL> """<STR_LIT>""" <EOL> return len ( remove_ansi_escape_sequences ( text ) ) <EOL> def format_table ( rows , min_spacing = <NUM_LIT:2> ) : <EOL> """<STR_LIT>""" <EOL> list_rows = [ r for r in rows if not isinstance ( r , basestring ) ] <EOL> if not list_rows : <EOL> return rows <EOL> widths = [ ] <EOL> for i in xrange ( len ( list_rows [ <NUM_LIT:0> ] ) ) : <EOL> widths . append ( max ( terminal_len ( r [ i ] ) for r in list_rows ) ) <EOL> expanded_rows = [ ] <EOL> for row in rows : <EOL> if row not in list_rows : <EOL> expanded_rows . append ( [ row ] ) <EOL> else : <EOL> expanded_row = [ ] <EOL> for i , cell in enumerate ( row ) : <EOL> if i == len ( row ) - <NUM_LIT:1> : <EOL> padding = '<STR_LIT>' <EOL> else : <EOL> padding = '<STR_LIT:U+0020>' * ( widths [ i ] - terminal_len ( cell ) ) <EOL> expanded_row . append ( cell + padding ) <EOL> expanded_rows . append ( expanded_row ) <EOL> return [ ( '<STR_LIT:U+0020>' * min_spacing ) . join ( r ) for r in expanded_rows ] <EOL> def deep_merge_dictionaries ( overrides , defaults ) : <EOL> """<STR_LIT>""" <EOL> result = copy . deepcopy ( defaults ) <EOL> stack = [ ( overrides , result ) ] <EOL> while stack : <EOL> source_dict , result_dict = stack . pop ( ) <EOL> for key , value in source_dict . items ( ) : <EOL> child = result_dict . setdefault ( key , { } ) <EOL> if isinstance ( value , dict ) and isinstance ( child , dict ) : <EOL> stack . append ( ( value , child ) ) <EOL> else : <EOL> result_dict [ key ] = value <EOL> return result <EOL> class ZookeeperPool ( object ) : <EOL> """<STR_LIT>""" <EOL> counter = <NUM_LIT:0> <EOL> zk = None <EOL> @ classmethod <EOL> def __enter__ ( cls ) : <EOL> if cls . zk is None : <EOL> cls . zk = KazooClient ( hosts = load_system_paasta_config ( ) . get_zk_hosts ( ) , read_only = True ) <EOL> cls . zk . start ( ) <EOL> cls . counter = cls . counter + <NUM_LIT:1> <EOL> return cls . zk <EOL> @ classmethod <EOL> def __exit__ ( cls , * args , ** kwargs ) : <EOL> cls . counter = cls . counter - <NUM_LIT:1> <EOL> if cls . counter == <NUM_LIT:0> : <EOL> cls . zk . stop ( ) <EOL> cls . zk . close ( ) <EOL> cls . zk = None </s>
