|
@@ -13,112 +13,291 @@
|
|
|
import json
|
|
|
from threading import Thread
|
|
|
import os
|
|
|
+import shutil
|
|
|
import subprocess
|
|
|
from monitor import ProcessCounter
|
|
|
-from utils.transport import QueueListener, QueueWriter
|
|
|
+from utils.transport import QueueListener, QueueWriter, QueueReader
|
|
|
+from utils.params import PARAMS
|
|
|
+from ngram import NGram as ng
|
|
|
class Actor(Thread):
|
|
|
- def __init__(self,config):
|
|
|
- Thread.__init__(self)
|
|
|
- self.config = config
|
|
|
- self.items = []
|
|
|
- self.__id = config['id']
|
|
|
- def getIdentifier(self):
|
|
|
- return self.__id
|
|
|
-
|
|
|
- def init(self,litems):
|
|
|
- self.items = litems
|
|
|
+ def __init__(self):
|
|
|
+ Thread.__init__(self)
|
|
|
+ pass
|
|
|
+ def getIdentifier(self):
|
|
|
+ return self.__class__.__name__.lower()
|
|
|
+ """
|
|
|
+ Initializing the class with configuration. The configuration will be specific to each subclass
|
|
|
+
|
|
|
+ """
|
|
|
+ def init(self,config,item=None):
|
|
|
+ self.config = config
|
|
|
+ self.item = item
|
|
|
def process(self,item):
|
|
|
pass
|
|
|
- def execute(self,cmd):
|
|
|
- stream = None
|
|
|
- try:
|
|
|
- print self.getIdentifier()
|
|
|
- print cmd
|
|
|
- handler = subprocess.Popen(cmd,shell=True,stdout=subprocess.PIPE)
|
|
|
- stream = handler.communicate()[0]
|
|
|
- except Exception,e:
|
|
|
- pass
|
|
|
- return stream
|
|
|
- def callback(self,channel,method,header,stream):
|
|
|
- print [self.getIdentifier(),stream]
|
|
|
- message = json.loads(stream)
|
|
|
- content = message['content']
|
|
|
- sender = message['from']
|
|
|
- if content.lower() == 'quit' :
|
|
|
- channel.close()
|
|
|
- print " *** ",self.getIdentifier()
|
|
|
- elif content.lower() == 'ping':
|
|
|
- self.post(to=sender,content="1")
|
|
|
- else:
|
|
|
- self.process(content)
|
|
|
- self.post(to=sender,content=content)
|
|
|
-
|
|
|
- #message = None
|
|
|
- #try:
|
|
|
- #message = json.loads(stream)
|
|
|
- #except Exception, e:
|
|
|
- #pass
|
|
|
- #if message is not None:
|
|
|
- #if 'id' in message :
|
|
|
- #if 'payload' in message:
|
|
|
- #self.execute(message['payload']
|
|
|
- """
|
|
|
- Sending a message to a queue with parameters to,from,content
|
|
|
- """
|
|
|
- def post(self,**args):
|
|
|
- to = args['to']
|
|
|
- content = args['content']
|
|
|
- message = {"from":self.getIdentifier(),"to":to,"content":content}
|
|
|
- host = self.config['api']
|
|
|
- uid = self.config['key']
|
|
|
- qid = to#self.config['id']
|
|
|
-
|
|
|
- qwriter = QueueWriter(host=host,uid=uid,qid=qid)
|
|
|
- qwriter.init(qid)
|
|
|
- qwriter.write(label=qid,row=content)
|
|
|
- #qwriter.close()
|
|
|
- pass
|
|
|
- def run(self):
|
|
|
- info = {}
|
|
|
- host = self.config['api']
|
|
|
- uid = self.config['key']
|
|
|
- qid = self.config['id']
|
|
|
-
|
|
|
- qlistener = QueueListener(qid=qid,uid=uid,host=host)
|
|
|
- qlistener.callback = self.callback
|
|
|
- qlistener.read()
|
|
|
- r = [self.process(item) for item in self.items]
|
|
|
+ def isValid(self,item):
|
|
|
+ return False
|
|
|
+
|
|
|
+ def execute(self,cmd):
|
|
|
+ stream = None
|
|
|
+ try:
|
|
|
+ subprocess.call (cmd,shell=False)
|
|
|
+ #stream = handler.communicate()[0]
|
|
|
+ except Exception,e:
|
|
|
+ pass
|
|
|
+
|
|
|
+ def run(self):
|
|
|
+ if self.item is not None:
|
|
|
+ self.process(self.item)
|
|
|
+ """
|
|
|
+ Sending a message to a queue with parameters to,from,content
|
|
|
+ """
|
|
|
+ def post(self,**args):
|
|
|
+ pass
|
|
|
+"""
|
|
|
+ This is designed to handle folders i.e cleaning/archiving the folders
|
|
|
+
|
|
|
+"""
|
|
|
+class Folders(Actor):
|
|
|
+ def init(self,config,item):
|
|
|
+ Actor.init(self,config,item)
|
|
|
+ self.lfolders = config['folders']
|
|
|
+ self.config = config['actions']['folders']
|
|
|
+ self.threshold = self.get_size(self.config['threshold'])
|
|
|
+ self.item = item
|
|
|
+
|
|
|
+ def archive(self,item):
|
|
|
+ """
|
|
|
+ This function will archive all files in a given folder
|
|
|
+ @pre : isValid
|
|
|
+ """
|
|
|
+ folder = item['label']
|
|
|
+ signature='-'.join([str(item['date']),str(item['count']),'-files'])
|
|
|
+ tarball=os.sep([folder,signature])
|
|
|
+ shutil.make_archive(tarball,'tar',folder)
|
|
|
+ self.clean(item)
|
|
|
+ #
|
|
|
+ # @TODO: The archive can be uploaded to the cloud or else where
|
|
|
+ # - This allows the submission of data to a processing engine if there ever were one
|
|
|
+ #
|
|
|
+ pass
|
|
|
+
|
|
|
+ def clean(self,item):
|
|
|
+ """
|
|
|
+ This function consists in deleting files from a given folder
|
|
|
+ """
|
|
|
+ rpath = item['label']
|
|
|
+ lists = os.listdir(item['label'])
|
|
|
+ for name in list() :
|
|
|
+ path = os.sep([item['label'],name])
|
|
|
+ if os.path.isdir(path) :
|
|
|
+ shutil.rmtree(path)
|
|
|
+ else:
|
|
|
+ os.remove(path)
|
|
|
+ #
|
|
|
+ #
|
|
|
+
|
|
|
+ def get_size(self,value):
|
|
|
+ units = {'MB':1000,'GB':1000000,'TB':1000000000} # converting to kb
|
|
|
+ key = set(unites) & set(re.split('(\d+)',value.upper()))
|
|
|
+ if len(key) == 0:
|
|
|
+ return -1
|
|
|
+ key = key.pop()
|
|
|
+ return float(value.upper().replace('MB','').strip()) * units[key]
|
|
|
+
|
|
|
+ def isvalid(self,item):
|
|
|
+ """
|
|
|
+ This function returns whether the following :
|
|
|
+ p : folder exists
|
|
|
+ q : has_reached threashold
|
|
|
+ """
|
|
|
+
|
|
|
+
|
|
|
+ p = os.path.exists(item['label']) and item['label'] in self.lfolders
|
|
|
+
|
|
|
+ q = self.get_size(item['size']) >= self.threshold
|
|
|
+ return p and q
|
|
|
+
|
|
|
+ def process(self,item):
|
|
|
+ if self.isValid(item) :
|
|
|
+
|
|
|
+ name = self.config['action']
|
|
|
+ stream = "".join([name,'(',json.dumps(item),')'])
|
|
|
+ eval(stream)
|
|
|
+
|
|
|
+
|
|
|
class Kill(Actor):
|
|
|
- def __init__(self,config):
|
|
|
- Actor.__init__(self,config)
|
|
|
+
|
|
|
+ def isValid(self,item):
|
|
|
+ return (item is not None) and (item in self.config)
|
|
|
def process(self,item):
|
|
|
- cmd = "".join(["ps -eo pid,command|grep ",item,'|grep -E "^ {0,1}[0-9]+" -o|xargs kill -9'])
|
|
|
- self.execute(cmd)
|
|
|
+ args = "".join(["-eo pid,command|grep ",item,'|grep -E "^ {0,1}[0-9]+" -o|xargs kill -9'])
|
|
|
+ self.execute(["ps",args])
|
|
|
#
|
|
|
# We need to make sure we can get assess the process on this server
|
|
|
#
|
|
|
+
|
|
|
class Start(Actor):
|
|
|
- def __init__(self,config):
|
|
|
- Actor.__init__(self,config)
|
|
|
- def process(self,item):
|
|
|
- path = item['path']
|
|
|
- args = item['args'] if 'args' in item else ''
|
|
|
- cmd = " ".join([path,args])
|
|
|
- self.execute(cmd)
|
|
|
-
|
|
|
+ def __init__(self):
|
|
|
+ Actor.__init__(self)
|
|
|
+ self.ng = None
|
|
|
+
|
|
|
+ def init(self,config,item):
|
|
|
+ Actor.init(self,config,item)
|
|
|
+ self.config = config['apps']
|
|
|
+ self.ng = ng(self.config.keys())
|
|
|
+
|
|
|
+ def isValid(self,name):
|
|
|
+ items = self.ng.search(name)
|
|
|
+ if len(items) == 0 :
|
|
|
+ return False
|
|
|
+ else:
|
|
|
+ return items[0][1] > 0.1
|
|
|
+
|
|
|
+ def process(self,row):
|
|
|
+ name = row['label']
|
|
|
+ items = self.ng.search(name)[0]
|
|
|
+ app = items[0]
|
|
|
+
|
|
|
+
|
|
|
+ args = self.config[app]
|
|
|
+ cmd = " ".join([app,args])
|
|
|
+
|
|
|
+ self.execute([app,args])
|
|
|
+"""
|
|
|
+ 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
|
|
|
+"""
|
|
|
+class Apps(Actor):
|
|
|
+ def __init__(self):
|
|
|
+ Actor.__init__(self)
|
|
|
+ self.crashes = []
|
|
|
+ self.running = []
|
|
|
+
|
|
|
+ def isValid(self,rows):
|
|
|
+ status = [row['status'] for row in rows]
|
|
|
+ return 'crash' in status
|
|
|
+
|
|
|
+ def classify(self,rows):
|
|
|
+ self.crashes = []
|
|
|
+ self.running = []
|
|
|
+ for row in rows:
|
|
|
+ if row['status'] == 'crash' :
|
|
|
+ self.crashes.append(row)
|
|
|
+ else:
|
|
|
+ self.running.append(row)
|
|
|
+ def reboot(self):
|
|
|
+ for row_run in self.running:
|
|
|
+ pass
|
|
|
+ def start(self):
|
|
|
+ for row_crash in self.crashes:
|
|
|
+ thread = Start()
|
|
|
+ thread.init(self.config,row_crash)
|
|
|
+ thread.daemon = True
|
|
|
+ thread.start()
|
|
|
+
|
|
|
+ def process(self,rows):
|
|
|
+ self.classify(rows)
|
|
|
+ if self.crashes :
|
|
|
+ self.start()
|
|
|
+ if self.running:
|
|
|
+ self.reboot()
|
|
|
+
|
|
|
+
|
|
|
+class Event(Thread):
|
|
|
+ def __init__(self,config):
|
|
|
+ pass
|
|
|
+ def run(self):
|
|
|
+"""
|
|
|
+ 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
|
|
|
+
|
|
|
+ The orchestrator is implemented using a simple iterator design-pattern
|
|
|
+ @TODO: action specifications should be provided remotely
|
|
|
+"""
|
|
|
+class Orchestrator(Actor):
|
|
|
+
|
|
|
+ def __init__(self,config=None):
|
|
|
+ Actor.__init__(self)
|
|
|
+ if config is None:
|
|
|
+ f = open(PARAMS['path'])
|
|
|
+ config = json.loads(f.read())
|
|
|
+ f.close()
|
|
|
+ self.config = config
|
|
|
+ Actor.__init__(self)
|
|
|
+
|
|
|
+ self.actors = {"apps":Apps(),"folders":Folders()}
|
|
|
+ self.is_master_node = False
|
|
|
+ self.items = []
|
|
|
+ #
|
|
|
+ # If the configuration only has id,key then this is NOT the maestro
|
|
|
+ #
|
|
|
+ host = config['api']
|
|
|
+ qid = config['id']
|
|
|
+ print "Initialized ***** ",self.getIdentifier(), " as ",config['id']
|
|
|
+
|
|
|
+ #
|
|
|
+ # This object will have to request for the configuration
|
|
|
+ #
|
|
|
+ #for id in config['actions'] :
|
|
|
+ #conf = config['actions'][id]
|
|
|
+ #item = eval("".join([id,"(",json.dumps(conf),")"]))
|
|
|
+ #self.actors[id.lower()] = item
|
|
|
+ """
|
|
|
+ This function is designed to provide the orchestrator a configuration
|
|
|
+ @pre
|
|
|
+ """
|
|
|
+ def init(self,config):
|
|
|
+
|
|
|
+ for id in config:
|
|
|
+
|
|
|
+ setup_info = config[id]
|
|
|
+ item = eval("".join([id,"(",json.dumps(setup_info),")"]))
|
|
|
+ self.actors[id.lower()] = item
|
|
|
+
|
|
|
+ def callback(self,channel,method,header,stream):
|
|
|
+
|
|
|
+ message = json.loads(stream)
|
|
|
+ if 'content' in message :
|
|
|
+ content = message['content']
|
|
|
+ print self.actors
|
|
|
+ to = message['to']
|
|
|
+ if isinstance(content,basestring) and content.lower() in ['quit'] or to=='quit':
|
|
|
+ if content.lower() == 'quit' or to == 'quit':
|
|
|
+ print '**** closing ',self.getIdentifier()
|
|
|
+ channel.close()
|
|
|
+ else:
|
|
|
+
|
|
|
+ id = to.lower()
|
|
|
+ actor = self.actors[id]
|
|
|
+
|
|
|
+ if actor is not None and actor.isValid(content) :
|
|
|
+ actor.init(self.config['actions'])
|
|
|
+ actor.process(content)
|
|
|
+ else:
|
|
|
+ content = {"status":"invalid","content":content}
|
|
|
+
|
|
|
+ #self.post(to=sender,content=content)
|
|
|
+
|
|
|
+ def run(self):
|
|
|
+
|
|
|
+ info = {}
|
|
|
+ host = self.config['api']
|
|
|
+ uid = self.config['key']
|
|
|
+ qid = self.config['id']
|
|
|
+
|
|
|
+ qlistener = QueueListener(qid=qid,uid=uid,host=host)
|
|
|
+ qlistener.callback = self.callback
|
|
|
+ qlistener.read()
|
|
|
+ r = [self.process(item) for item in self.items]
|
|
|
+
|
|
|
"""
|
|
|
This class is designed to send a message to a given AMQP enpoint
|
|
|
The AMQP endpoint is implemented by QueueWriter class
|
|
|
"""
|
|
|
-class Alert(Actor):
|
|
|
- def process(self,item):
|
|
|
- pass
|
|
|
-
|
|
|
-
|
|
|
-config = {"id":"demo","key":"[0v8]-247&7!v3","api":"localhost"}
|
|
|
-actor = Kill(config)
|
|
|
-actor.start()
|
|
|
+# class Alert(Actor):
|
|
|
+# def process(self,item):
|
|
|
+# pass
|
|
|
|
|
|
-config = {"id":"demo-100","key":"[0v8]-247&7!v3","api":"localhost"}
|
|
|
-actor_1 = Kill(config)
|
|
|
-actor_1.start()
|
|
|
+if __name__ == '__main__':
|
|
|
+ thread = Orchestrator()
|
|
|
+ thread.start()
|