黑松山资源网 Design By www.paidiu.com
salt分发后,主动将已完成的任务数据推送到redis中,使用redis的生产者模式,进行消息传送
#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 队列监听器 ''' 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() '''
以上这篇python 监听salt job状态,并任务数据推送到redis中的方法就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持。
黑松山资源网 Design By www.paidiu.com
广告合作:本站广告合作请联系QQ:858582 申请时备注:广告合作(否则不回)
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
黑松山资源网 Design By www.paidiu.com
暂无评论...
更新日志
2024年10月07日
2024年10月07日
- 群星.1991-音乐工厂1·皇后大道东【音乐工厂】【WAV+CUE】
- 群星.1992-音乐工厂2·首都【音乐工厂】【WAV+CUE】
- 群星.1994-音乐工厂3·儿童乐园【音乐工厂】【WAV+CUE】
- 胡鸿钧.2021-ex:CHANGE【星梦娱乐】【WAV+CUE】
- 班得瑞原装进口《第一张新世纪专辑:仙境》1CD[APE/CUE分轨][292.3MB]
- 伍佰《摇滚教父 伍佰黄金精选 2CD》[WAV+CUE][990MB]
- 班得瑞原装进口《第二张新世纪专辑:寂静山林》1CD[APE/CUE分轨][327.1MB]
- Beyond.1989-真的见证(2024环球MQA-UHQCD限量版)【环球】【WAV+CUE】
- 陈怡婷.2024-予你的心谁来赔【福茂】【FLAC分轨】
- 陈冠希.2003-HITS.OR.MISSES【英皇娱乐】【WAV+CUE】
- 班得瑞原装进口《第三张新世纪专辑:春野》1CD[APE/CUE分轨][278.9MB]
- 班得瑞原装进口《第四张新世纪专辑:蓝色天际》1CD[APE/CUE分轨][284.8MB]
- 班得瑞原装进口《第五张新世纪专辑:迷雾森林+EP》1CD[APE/CUE分轨][334.2MB]
- 田震.1997-顺其自然(日版)【红星生产社】【WAV+CUE】
- 小虫.1996-想得太美【滚石】【WAV+CUE】