Add 'src/utils/agents/data-manager.py'

data-collector manager

src/utils/agents/data-manager.py

@@ -0,0 +1,87 @@
+from threading import Thread, RLock
+
+import os
+import json
+import time
+from datetime import datetime
+from utils.transport import *
+import monitor
+
+class Manager(Thread) :
+	"""
+		delay : <value>
+		limit : <value>
+		scope : apps,folders,learner,sandbox
+	"""
+	def __init__(self):
+		Thread.__init__(self)
+		self.lock 	= RLock()
+		self.factory	= DataSourceFactory()
+	def init(self,args) :
+		node,pool,config
+		self.id		= args['node']
+		self.pool 	= args['pool']
+		self.config	= args['config']
+		self.key	= args['key']
+
+		self.status()	#-- Initializing status information
+	def status(self) :
+		"""
+			This method inspect the plans for the current account and makes sure it can/should proceed
+			The user must be subscribed and to the service otherwise this is not going to work
+		"""
+		url="https://the-phi.com/store/status/monitor"
+		r = requests.post(url,headers={"uid":self.key})
+		plans = json.loads(r.text)
+
+		meta =  [item['metadata'] for item in plans if item['status']=='active' ]
+		if len(meta) > 0 :
+			self.DELAY = 60* max([ int(item['delay']) for item in meta if ])
+			self.LIMIT = max([ int(item['limit']) for item in meta if ])
+		else:
+			self.DELAY = -1
+			self.LIMIT = -1
+		scope = []
+		[ scope += item['scope'].split(',') for item in meta ]
+		names = [ for agent in self.pool if agent.getName() in scope]
+		return meta
+
+	def isvalid(self):
+		self.status()
+		return self.DELAY > -1 and self.LIMIT > -1
+	def run(self):
+		#DELAY=35*60 #- 35 Minutes
+		#LIMIT=1000
+		COUNT = 0
+		COUNT_STOP = int(24*60/ self.DELAY)
+		print COUNT_STOP
+		write_class 	= self.config['store']['class']['write']
+		read_args	= self.config['store']['args']
+		
+		while True :
+			COUNT += 1
+			if COUNT > COUNT_STOP :
+				if self.isvalid() :
+					COUNT = 0
+				else:
+					break
+			for agent in self.pool :
+				
+				data	= agent.composite()
+				label	= agent.getName()
+				node = '@'.join([label,self.id])
+				row	= {}
+				if label == 'folders':
+					row = [ dict({"id":self.id}, **_row) for _row in data]					
+					
+				else:
+					label = id
+					row = data
+
+				self.lock.acquire()
+				store = self.factory.instance(type=write_class,args=read_args)
+				store.flush(size=self.LIMIT)
+				store.write(label=label,row=row)
+				self.lock.release()
+			time.sleep(self.DELAY)
+