溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊(cè)×
其他方式登錄
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶(hù)服務(wù)條款》

python 監(jiān)聽(tīng)salt job狀態(tài),并任務(wù)數(shù)據(jù)推送到redis中的方法

發(fā)布時(shí)間:2020-09-08 20:44:53 來(lái)源:腳本之家 閱讀:190 作者:云中不知人 欄目:開(kāi)發(fā)技術(shù)

salt分發(fā)后,主動(dòng)將已完成的任務(wù)數(shù)據(jù)推送到redis中,使用redis的生產(chǎn)者模式,進(jìn)行消息傳送

#coding=utf-8
import fnmatch,json,logging
import salt.config
import salt.utils.event
from salt.utils.redis import RedisPool
import sys,os,datetime,random
import multiprocessing,threading
from joi.utils.gobsAPI import PostWeb
logger = 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 隊(duì)列監(jiān)聽(tīng)器
	'''
	def __init__(self,r_conn):
		self.r_conn = r_conn #redis 連接實(shí)例
		self.task_queue = 'task:prod:queue' #任務(wù)消息隊(duì)列
	def listen_task(self):
		'''
		監(jiān)聽(tīng)主函數(shù)
		'''
		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):
		'''
		執(zhí)行操作函數(shù)
		'''
		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('獲取緩存數(shù)據(jù):%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('發(fā)送任務(wù)數(shù)據(jù)到后端...')
		logger.info(task_data)
		if task_data:
			p = PostWeb('/jgapi/verify',task_data,'pushFlowTaskData')
			result = p.postRes()
			print result
			if result['code']:
				logger.info('發(fā)送成功!')
				return True
			else:
				logger.error('發(fā)送失??!')
				return False
		else:
			return True
	def setTaskState(self,jid,state,message=''):
		logger.info('到后端設(shè)置任務(wù)【%s】狀態(tài)' % str(jid))
		p = PostWeb('/jgapi/verify',{'code':jid,'state':'success','message':message},'setTaskState')
		result = p.postRes()
		if result['code']:
			logger.info('設(shè)置任務(wù)【%s】狀態(tài)成功!' % str(jid))
			return True,result
		else:
			logger.error('設(shè)置任務(wù)【%s】狀態(tài)失敗!' % str(jid))
			return result		
def salt_job_listener():
	'''
	salt job 監(jiān)聽(tīng)器
	'''
	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("新增數(shù)據(jù):%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('新增消息到隊(duì)列:pushTaskData')
						r_conn.lpush('task:prod:queue',json.dumps({'jid':ret['data']['jid'],'type':'pushTaskData'}))
				else:
					logger.info('任務(wù){(diào)0}完成,發(fā)送剩下的數(shù)據(jù)到后端...'.format(task_key))
					logger.info('新增消息到隊(duì)列: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()
	'''

以上這篇python 監(jiān)聽(tīng)salt job狀態(tài),并任務(wù)數(shù)據(jù)推送到redis中的方法就是小編分享給大家的全部?jī)?nèi)容了,希望能給大家一個(gè)參考,也希望大家多多支持億速云。

向AI問(wèn)一下細(xì)節(jié)

免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀(guān)點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI