library
/
smart-top
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

CO - actor refactor for killing and starting an application

Browse Source
community
Steve L. Nyemba 7 years ago
parent 31ff495f74
commit 2a3a0fcb96
1 changed files with 228 additions and 115 deletions
Show Stats Download Patch File Download Diff File

343
src/utils/agents/actor.py
Unescape View File

@ -19,19 +19,36 @@ from monitor import ProcessCounter
from utils.transport import QueueListener, QueueWriter, QueueReader from utils.transport import QueueListener, QueueWriter, QueueReader
from utils.params import PARAMS from utils.params import PARAMS
from ngram import NGram as ng from ngram import NGram as ng
class Actor(Thread): import smtplib
def __init__(self): from email.mime.multipart import MIMEMultipart
Thread.__init__(self) from email.mime.text import MIMEText
class Actor():
@staticmethod
def instance(name,args):
"""
This function is a singleton that acts as a factory object for all the instances of this subclass
@param name name of the class to instantiate
@param args arguments to be passed in {configuration}
"""
o = None
try:
o = eval("".join([name,"()"]))
o.init(args)
except Exception,e:
print e
return o
def __init__(self):
pass pass
def getIdentifier(self): def getIdentifier(self):
return self.__class__.__name__.lower() return self.__class__.__name__.lower()
""" """
Initializing the class with configuration. The configuration will be specific to each subclass Initializing the class with configuration. The configuration will be specific to each subclass
@param args arguments the class needs to be configured
""" """
def init(self,config,item=None): def init(self,args):
self.config = config self.config = args
self.item = item
def process(self,item): def process(self,item):
pass pass
def isValid(self,item): def isValid(self,item):
@ -45,31 +62,123 @@ class Actor(Thread):
except Exception,e: except Exception,e:
pass pass
def run(self): # def run(self):
if self.item is not None: # if self.item is not None:
self.process(self.item) # self.process(self.item)
""" """
Sending a message to a queue with parameters to,from,content Sending a message to a queue with parameters to,from,content
""" """
def post(self,**args): def post(self,**args):
pass pass
""" class Apps(Actor) :
This is designed to handle folders i.e cleaning/archiving the folders """
This class is designed to handle application, restart, if need be.
conf{app-name:{args}}
"""
def __init__(self):
Actor.__init__(self)
self.ng = None
def init(self,config) :
"""
This function will initialize the the actor with applications and associated arguments
@param args {"apps_o":"","app_x":params}
"""
Actor.init(self,config)
self.ng = ng(self.config.keys())
"""
def can_start(self,name):
"""
This function is intended to determine if it is possible to boot an application
"""
items = self.ng.search(name)
if len(items) == 0 :
return False
else:
return items[0][1] > 0.01
def startup(self,name) :
"""
This function is intended to start a program given the configuration
"""
items = self.ng.search(name)[0]
app = items[0]
args = self.config[app]
cmd = " ".join([app,args])
self.execute([app,args])
def kill(self,name) :
args = "".join(["-eo pid,command|grep -Ei ",name.lower(),'|grep -E "^ {0,1}[0-9]+" -o|xargs kill -9'])
self.execute(["ps",args])
def analyze(self,logs) :
"""
This function is designed to analyze a few logs and take appropriate action
@param logs logs of application/process data; folder analysis or sandbox analysis
"""
for item in self.logs :
name = item['label']
if self.can_start(name) :
self.startup(name)
#
#
class Mailer (Actor):
"""
conf = {uid:<account>,host:<host>,port:<port>,password:<password>}
"""
def __init__(self,conf) :
self.uid = conf['uid']
try:
self.handler = smtplib.SMTP_SSL(conf['host'],conf['port'])
r = self.handler.login(self.uid,conf['password'])
#
# @TODO: Check the status of the authentication
# If not authenticated the preconditions have failed
#
except Exception,e:
print e
self.handler = None
pass
def send(self,**args) :
subject = args['subject']
message = args['message']
to = args['to']
if '<' in message and '>' in message :
message = MIMEText(message,'html')
else:
message = MIMEText(message,'plain')
message['From'] = self.uid
message['To'] = to
message['Subject'] = subject
return self.handler.sendmail(self.uid,to,message.as_string())
def close(self):
self.handler.quit()
class Folders(Actor): class Folders(Actor):
def init(self,config,item): """
This is designed to handle folders i.e cleaning/archiving the folders
if the user does NOT have any keys to cloud-view than she will not be able to archive
"""
def init(self,**args):
Actor.init(self,config,item) Actor.init(self,config,item)
self.lfolders = config['folders'] self.lfolders = args['folders'] #config['folders']
self.config = config['actions']['folders'] self.action = args['action'] #{clear,archive} config['actions']['folders']
self.threshold = self.get_size(self.config['threshold']) self.threshold = self.get_size( args['threshold']) #self.config['threshold'])
self.item = item self.item = item
def archive(self,item): def archive(self,item):
""" """
This function will archive all files in a given folder This function will archive all files in a given folder
@pre : isValid @pre : isValid
""" """
folder = item['label'] folder = item['label']
signature='-'.join([str(item['date']),str(item['count']),'-files']) signature='-'.join([str(item['date']),str(item['count']),'-files'])
tarball=os.sep([folder,signature]) tarball=os.sep([folder,signature])
@ -82,12 +191,12 @@ class Folders(Actor):
pass pass
def clean(self,item): def clean(self,item):
""" """
This function consists in deleting files from a given folder This function consists in deleting files from a given folder
""" """
rpath = item['label'] rpath = item['label']
lists = os.listdir(item['label']) files = os.listdir(item['label'])
for name in list() : for name in list(files) :
path = os.sep([item['label'],name]) path = os.sep([item['label'],name])
if os.path.isdir(path) : if os.path.isdir(path) :
shutil.rmtree(path) shutil.rmtree(path)
@ -105,123 +214,127 @@ class Folders(Actor):
return float(value.upper().replace('MB','').strip()) * units[key] return float(value.upper().replace('MB','').strip()) * units[key]
def isvalid(self,item): def isvalid(self,item):
""" """
This function returns whether the following : This function returns whether the following :
p : folder exists p : folder exists
q : has_reached threashold q : has_reached threashold
""" """
p = os.path.exists(item['label']) and item['label'] in self.lfolders p = os.path.exists(item['label']) and item['label'] in self.lfolders
q = self.get_size(item['size']) >= self.threshold q = self.get_size(item['size']) >= self.threshold
return p and q return p and q
def process(self,item): def analyze(self,logs):
if self.isValid(item) : r = {'clean':self.clean,'archive':self.archive}
for item in logs :
name = self.config['action'] if self.isValid(item) :
stream = "".join([name,'(',json.dumps(item),')'])
eval(stream) id = self.config['action'].strip()
pointer = r[id]
pointer (item)
class Kill(Actor): # class Kill(Actor):
def isValid(self,item): # def isValid(self,item):
return (item is not None) and (item in self.config) # return (item is not None) and (item in self.config)
def process(self,item): # def process(self,item):
args = "".join(["-eo pid,command|grep ",item,'|grep -E "^ {0,1}[0-9]+" -o|xargs kill -9']) # args = "".join(["-eo pid,command|grep ",item,'|grep -E "^ {0,1}[0-9]+" -o|xargs kill -9'])
self.execute(["ps",args]) # self.execute(["ps",args])
# # #
# We need to make sure we can get assess the process on this server # # We need to make sure we can get assess the process on this server
# # #
class Start(Actor): # class Start(Actor):
def __init__(self): # def __init__(self):
Actor.__init__(self) # Actor.__init__(self)
self.ng = None # self.ng = None
def init(self,config,item): # def init(self,config,item):
Actor.init(self,config,item) # Actor.init(self,config,item)
self.config = config['apps'] # self.config = config['apps']
self.ng = ng(self.config.keys()) # self.ng = ng(self.config.keys())
def isValid(self,name): # def isValid(self,name):
items = self.ng.search(name) # items = self.ng.search(name)
if len(items) == 0 : # if len(items) == 0 :
return False # return False
else: # else:
return items[0][1] > 0.1 # return items[0][1] > 0.1
def process(self,row): # def process(self,row):
name = row['label'] # name = row['label']
items = self.ng.search(name)[0] # items = self.ng.search(name)[0]
app = items[0] # app = items[0]
args = self.config[app] # args = self.config[app]
cmd = " ".join([app,args]) # cmd = " ".join([app,args])
self.execute([app,args]) # self.execute([app,args])
""" # """
This class is designed to handle applications i.e start/stopping applications # This class is designed to handle applications i.e start/stopping applications
@TODO: Assess if a reboot is required, by looking at the variance/anomaly detection # @TODO: Assess if a reboot is required, by looking at the variance/anomaly detection
""" # """
class Apps(Actor): # class Apps(Actor):
def __init__(self): # def __init__(self):
Actor.__init__(self) # Actor.__init__(self)
self.crashes = [] # self.crashes = []
self.running = [] # self.running = []
def isValid(self,rows): # def isValid(self,rows):
status = [row['status'] for row in rows] # status = [row['status'] for row in rows]
return 'crash' in status # return 'crash' in status
def classify(self,rows): # def classify(self,rows):
self.crashes = [] # self.crashes = []
self.running = [] # self.running = []
for row in rows: # for row in rows:
if row['status'] == 'crash' : # if row['status'] == 'crash' :
self.crashes.append(row) # self.crashes.append(row)
else: # else:
self.running.append(row) # self.running.append(row)
def reboot(self): # def reboot(self):
for row_run in self.running: # for row_run in self.running:
pass # pass
def start(self): # def start(self):
for row_crash in self.crashes: # for row_crash in self.crashes:
thread = Start() # thread = Start()
thread.init(self.config,row_crash) # thread.init(self.config,row_crash)
thread.daemon = True # thread.daemon = True
thread.start() # thread.start()
def process(self,rows): # def process(self,rows):
self.classify(rows) # self.classify(rows)
if self.crashes : # if self.crashes :
self.start() # self.start()
if self.running: # if self.running:
self.reboot() # self.reboot()
class Event(Thread): class Event(Thread):
def __init__(self,config): def __init__(self,config):
pass pass
def run(self): def run(self):
""" """
The orchestrator class is designed to aggregate actions and communicate back to the caller The orchestrator class is designed to aggregate actions and communicate back to the caller
Mesage passing is structured as follows {from,to,content} The content is designed to be understood by the actor Mesage passing is structured as follows {from,to,content} The content is designed to be understood by the actor
The orchestrator is implemented using a simple iterator design-pattern The orchestrator is implemented using a simple iterator design-pattern
@TODO: action specifications should be provided remotely @TODO: action specifications should be provided remotely
""" """
pass
class Orchestrator(Actor): class Orchestrator(Actor):
def __init__(self,config=None): def __init__(self,config=None):
Actor.__init__(self) Actor.__init__(self)
if config is None: # if config is None:
f = open(PARAMS['path']) # f = open(PARAMS['path'])
config = json.loads(f.read()) # config = json.loads(f.read())
f.close() # f.close()
self.config = config self.config = config
Actor.__init__(self) Actor.__init__(self)

Write Preview
Loading…
Cancel
Save