python 监听salt job状态,并任务数据推送到redis中的方法
作者:云中不知人 发布时间:2022-09-14 05:19:47
标签:python,salt,job,redis
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()
'''
来源:https://blog.csdn.net/u011085172/article/details/81228450


猜你喜欢
- with 语句是从 Python 2.5 开始引入的一种与异常处理相关的功能(2.5 版本中要通过 from __future__ impo
- 一、总结说明Windows环境安装:paramunittest cmd输入命令:pip install paramunittest总结说明:
- 最近研究研究了css,少不了去网上分析一下别人的网页, 但很多网站的css文件都是要么写在一行,要么一个换行都没有,看起来极其痛苦,所以写一
- 之前一直对于python类的继承机制认知的比较混乱,今天学习记录一下。(1)首先使用直接继承的方式class parent():  
- 目录1、Unittest为Python内嵌的测试框架,不需要特殊配置2、编写规范总结1、Unittest为Python内嵌的测试框架,不需要
- 没怎么用过这个新特性,其实也不算新啦,试试吧,现在静态类的继承很方便了<?phpclass A { protected static
- 目录process模块1、在python中启动一个子进程2、给子进程传递参数3、同时开多个子进程4、join的用法5、多进程之间的数据是否隔
- 本文将介绍在InterDev中实现网上商店购物车功能的方法,具体步骤如下:一、 数据库结构:产品数据表(Products): 存放产品信息产
- 前言大家都看过彩带飘落吧?这个在比较喜庆的场合是很常见的:还有“跑马灯”效果,听起来很陌生,其实很常见,下面的就是:来源:https://w
- 介绍我编写了一个快速且带有斑点的python脚本,以可视化nmap和masscan的结果。它通过解析来自扫描的XML日志并生成所扫描IP范围
- 相比于2018年,在ICLR2019提交论文中,提及不同框架的论文数量发生了极大变化,网友发现,提及tensorflow的论文数量从2018
- 小的时候大家应该都玩过万花尺,将笔尖放置万花尺内不停的转动,便可以画出一幅精致的线稿图,现在我们不用万花尺,我们通过Python也能绘制出万
- 我们可以利用windows提供的api函数来实现对系统键盘事件和鼠标事件的监听,主要利用的是SetWindowsHookEx函数,这个函数可
- 前言自从学过存储过程后,就再也没有碰过存储过程,这是毕业后写的第一个存储过程。因为项目里设备的种类比较多,分别存在不同的数据表中,java中
- 在windows平台上使用pyhton编写语音识别程序需要用到speech模块,speech模块支持的主要功能有:文本合成语音,将键盘输入的
- 前言作为一个数据分析师,应该信奉一句话——“一图胜千言”。不过这里要说的并不是数据可视化,而是一款全民向的产品形态——表情包!!!!表情包不
- 1:下载FCKEditor编辑器下载下来后解压到你网站的目录,最好就放在根目录下,文件夹名字就用FCKEditor;这里可以随便自己喜好,但
- python读取和保存图片5种方法对比python中对象之间的赋值是按引用传递的,如果需要拷贝对象,需要用到标准库中的copy模块方法一:利
- 本文实例讲述了python实现的简单FTP上传下载文件的方法。分享给大家供大家参考。具体如下:python本身自带一个FTP模块,可以实现上
- 刚入手vue,有好多的疑问,目前遇到的是vue项目启动之后,输入http://ip:port,我发现浏览器里面显示的地址栏竟然是http:/