python 监听salt job状态,并任务数据推送到redis中的方法

#coding=utf-8import fnmatch,json,loggingimport salt.configimport salt.utils.eventfrom salt.utils.redis import RedisPoolimport sys,os,datetime,randomimport multiprocessing,threadingfrom joi.utils.gobsAPI import PostWeblogger = logging.getLogger(__name__)opts = salt.config.client_config('/data/salt/saltstack/etc/salt/master')r_conn = RedisPool(opts.get('redis_db')).getConn()lock = threading.Lock()class RedisQueueDaemon(object):	'''	redis 队列监听器	'''	def __init__(self,r_conn):		self.r_conn = r_conn #redis 连接实例		self.task_queue = 'task:prod:queue' #任务消息队列	def listen_task(self):		'''		监听主函数		'''		while True:				queue_item = self.r_conn.blpop(self.task_queue,0)[1]				print "queue get",queue_item				#self.run_task(queue_item)				t = threading.Thread(target=self.run_task,args=(queue_item,))				t.start()	def run_task(self,info):		'''		执行操作函数		'''		lock.acquire()		info = json.loads(info)		if info['type'] == 'pushTaskData':			task_data = self.getTaskData(info['jid'])			task_data = json.loads(task_data) if task_data else []			logger.info('获取缓存数据：%s' % task_data)			if task_data:				if self.sendTaskData2bs(task_data):					task_data = []			self.setTaskData(info['jid'], task_data)		elif info['type'] == 'setTaskState':			self.setTaskState(info['jid'],info['state'],info['message'])		elif info['type'] == 'setTaskData':			self.setTaskData(info['jid'], info['data'])		lock.release()	def getTaskData(self,jid):		return self.r_conn.hget('task:'+jid,'data')	def setTaskData(self,jid,data):		self.r_conn.hset('task:'+jid,'data',json.dumps(data))	def sendTaskData2bs(self,task_data):		logger.info('发送任务数据到后端...')		logger.info(task_data)		if task_data:			p = PostWeb('/jgapi/verify',task_data,'pushFlowTaskData')			result = p.postRes()			print result			if result['code']:				logger.info('发送成功！')				return True			else:				logger.error('发送失败！')				return False		else:			return True	def setTaskState(self,jid,state,message=''):		logger.info('到后端设置任务【%s】状态' % str(jid))		p = PostWeb('/jgapi/verify',{'code':jid,'state':'success','message':message},'setTaskState')		result = p.postRes()		if result['code']:			logger.info('设置任务【%s】状态成功!' % str(jid))			return True,result		else:			logger.error('设置任务【%s】状态失败!' % str(jid))			return result		def salt_job_listener():	'''	salt job 监听器	'''	sevent = salt.utils.event.get_event(			'master',			sock_dir=opts['sock_dir'],			transport=opts['transport'],			opts=opts)		while True:		ret = sevent.get_event(full=True)		if ret is None:			continue		if fnmatch.fnmatch(ret['tag'], 'salt/job/*/ret/*'):			task_key = 'task:'+ret['data']['jid']			task_state = r_conn.hget(task_key,'state')			task_data = r_conn.hget(task_key,'data')			if task_state:				jid_data = {					'code':ret['data']['jid'],					'project_id':settings.SALT_MASTER_OPTS['project_id'],					'serverip':ret['data']['id'],					'returns':ret['data']['return'],					'name':ret['data']['id'],					'state':'success' if ret['data']['success'] else 'failed',				}				task_data = json.loads(task_data) if task_data else []				task_data.append(jid_data)				logger.info("新增数据：%s" % json.dumps(task_data))				r_conn.lpush('task:prod:queue',json.dumps({'type':'setTaskData','jid':ret['data']['jid'],'data':task_data}))				#r_conn.hset(task_key,'data',json.dumps(task_data))										if task_state == 'running':					if len(task_data)>=1:						logger.info('新增消息到队列：pushTaskData')						r_conn.lpush('task:prod:queue',json.dumps({'jid':ret['data']['jid'],'type':'pushTaskData'}))				else:					logger.info('任务{0}完成，发送剩下的数据到后端...'.format(task_key))					logger.info('新增消息到队列：pushTaskData')					r_conn.lpush('task:prod:queue',json.dumps({'jid':ret['data']['jid'],'type':'pushTaskData'}))								print datetime.datetime.now() def run():	print 'start redis product queue listerner...'	logger.info('start redis product queue listerner...')	multiprocessing.Process(target=RedisQueueDaemon(r_conn).listen_task,args=()).start()	print 'start salt job listerner...'	logger.info('start salt job listerner...')	multiprocessing.Process(target=salt_job_listener,args=()).start() 	'''	p=multiprocessing.Pool(2)	print 'start redis product queue listerner...'	p.apply_async(redis_queue_listenr,())	print 'start salt job listerner...'	p.apply_async(salt_job_listener,())	p.close()	p.join()	'''