<s> import json <EOL> import logging <EOL> import gevent <EOL> import os <EOL> import psutil <EOL> import six <EOL> from gevent import subprocess <EOL> from six . moves import range <EOL> from lymph . core . interfaces import Interface <EOL> from lymph . core . monitoring . metrics import Generator <EOL> from lymph . utils . sockets import create_socket <EOL> logger = logging . getLogger ( __name__ ) <EOL> class Process ( object ) : <EOL> def __init__ ( self , cmd , env = None , service_type = '<STR_LIT>' ) : <EOL> self . cmd = cmd <EOL> self . env = env <EOL> self . service_type = service_type <EOL> self . _process = None <EOL> self . _popen = None <EOL> def is_running ( self ) : <EOL> return self . _process and self . _process . is_running ( ) <EOL> def start ( self ) : <EOL> self . _popen = subprocess . Popen ( <EOL> self . cmd , env = self . env , close_fds = False ) <EOL> self . _process = psutil . Process ( self . _popen . pid ) <EOL> self . _process . cpu_percent ( ) <EOL> def stop ( self , ** kwargs ) : <EOL> signalnum = kwargs . get ( '<STR_LIT>' ) <EOL> try : <EOL> if signalnum : <EOL> self . _process . send_signal ( signalnum ) <EOL> else : <EOL> self . _process . terminate ( ) <EOL> self . _process . wait ( ) <EOL> except psutil . NoSuchProcess : <EOL> pass <EOL> def restart ( self ) : <EOL> print ( "<STR_LIT>" % self ) <EOL> self . stop ( ) <EOL> self . start ( ) <EOL> def get_metrics ( self ) : <EOL> if not self . is_running ( ) : <EOL> return <EOL> tags = { '<STR_LIT>' : self . service_type } <EOL> try : <EOL> memory = self . _process . memory_info ( ) <EOL> yield '<STR_LIT>' , memory . rss , tags <EOL> yield '<STR_LIT>' , memory . vms , tags <EOL> yield '<STR_LIT>' , self . _process . cpu_percent ( ) , tags <EOL> except psutil . NoSuchProcess : <EOL> pass <EOL> class Node ( Interface ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( Node , self ) . __init__ ( * args , ** kwargs ) <EOL> self . sockets = { } <EOL> self . processes = [ ] <EOL> self . running = False <EOL> self . _sockets = [ ] <EOL> self . _services = [ ] <EOL> def apply_config ( self , config ) : <EOL> for name , c in six . iteritems ( config . get ( '<STR_LIT>' , { } ) ) : <EOL> self . _services . append ( ( name , c . get ( '<STR_LIT>' ) , c . get ( '<STR_LIT>' , <NUM_LIT:1> ) ) ) <EOL> socket_config = config . get_raw ( '<STR_LIT>' , ( ) ) <EOL> if isinstance ( socket_config , dict ) : <EOL> socket_config = socket_config . values ( ) <EOL> for c in socket_config : <EOL> self . _sockets . append ( ( c . get ( '<STR_LIT:host>' ) , c . get ( '<STR_LIT:port>' ) ) ) <EOL> def on_start ( self ) : <EOL> self . create_shared_sockets ( ) <EOL> self . running = True <EOL> shared_fds = json . dumps ( { port : s . fileno ( ) for port , s in six . iteritems ( self . sockets ) } ) <EOL> for service_type , cmd , num in self . _services : <EOL> env = os . environ . copy ( ) <EOL> env . update ( { <EOL> '<STR_LIT>' : self . container . endpoint , <EOL> '<STR_LIT>' : self . container . monitor . endpoint , <EOL> '<STR_LIT>' : self . container . server . ip , <EOL> '<STR_LIT>' : shared_fds , <EOL> '<STR_LIT>' : service_type , <EOL> } ) <EOL> for i in range ( num ) : <EOL> p = Process ( cmd . split ( '<STR_LIT:U+0020>' ) , env = env , service_type = service_type ) <EOL> self . processes . append ( p ) <EOL> self . metrics . add ( Generator ( p . get_metrics ) ) <EOL> logger . info ( '<STR_LIT>' , cmd ) <EOL> p . start ( ) <EOL> self . container . spawn ( self . watch_processes ) <EOL> def on_stop ( self , ** kwargs ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> self . running = False <EOL> for p in self . processes : <EOL> p . stop ( ** kwargs ) <EOL> super ( Node , self ) . on_stop ( ** kwargs ) <EOL> def create_shared_sockets ( self ) : <EOL> for host , port in self . _sockets : <EOL> sock = create_socket ( <EOL> '<STR_LIT>' % ( host or self . container . server . ip , port ) , inheritable = True ) <EOL> self . sockets [ port ] = sock <EOL> def watch_processes ( self ) : <EOL> while True : <EOL> for process in self . processes : <EOL> try : <EOL> status = process . _process . status ( ) <EOL> except psutil . NoSuchProcess : <EOL> if self . running : <EOL> process . start ( ) <EOL> continue <EOL> if status in ( psutil . STATUS_ZOMBIE , psutil . STATUS_DEAD ) : <EOL> if self . running : <EOL> process . restart ( ) <EOL> gevent . sleep ( <NUM_LIT:1> ) </s>
