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()
	'''

以上这篇python 监听salt job状态,并任务数据推送到redis中的方法就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持我们。

(0)

相关推荐

  • 通过Python使用saltstack生成服务器资产清单

    SaltStack是一个服务器基础架构集中化管理平台,具备配置管理.远程执行.监控等功能,一般可以理解为简化版的puppet和加强版的func.SaltStack基于Python语言实现,结合轻量级消息队列(ZeroMQ)与Python第三方模块(Pyzmq.PyCrypto.Pyjinjia2.python-msgpack和PyYAML等)构建. 通过部署SaltStack环境,我们可以在成千上万台服务器上做到批量执行命令,根据不同业务特性进行配置集中化管理.分发文件.采集服务器数据.操作系统

  • 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.util

  • Android 监听软键盘状态的实例详解

    Android 监听软键盘状态的实例详解 近日遇到要检测软键盘是否显示或隐藏的问题,搜了一下网上,最后找到一个很简单的,记录一下. activityRoot是activity的根view,就是xml里面的第一个view,给它设置一个id. final View activityRootView = findViewById(R.id.activityRoot); activityRootView.getViewTreeObserver().addOnGlobalLayoutListener(ne

  • Android获取监听SD卡状态

    项目简介: 该项目监听SD卡状态 详细介绍: 部署该应用后,当SD卡状态发送改变时,系统会发出吐司通知用户 该应用涉及到的知识有: 1.Android的四大组件之一:广播接收者的使用 2.SD卡状态的广播 注意: 1.只有在 2.x 模拟器上才能对SD卡进行操作,所以该应用应当部署在 2.x 的模拟器上 2.一个广播接收者是可以接受多个广播的,直接添加多个action就可以了 3.SD卡不同的状态有不同的广播,主要有三种: <action android:name="android.int

  • Android 利用广播监听usb连接状态(变化情况)

    废话不多说了,直接给大家贴代码了,具体代码如下所示: package com.lgs.test.testcode.receiver; import android.content.BroadcastReceiver; import android.content.Context; import android.content.Intent; import android.content.IntentFilter; import android.widget.Toast; /** * Create

  • Android编程监听网络连接状态改变的方法

    本文实例讲述了Android编程监听网络连接状态改变的方法.分享给大家供大家参考,具体如下: BroadcastReceiver public class MyReceiver extends BroadcastReceiver { @Override public void onReceive(Context context, Intent intent) { // TODO Auto-generated method stub //Toast.makeText(context, intent

  • Android监听手机电话状态与发送邮件通知来电号码的方法(基于PhoneStateListene实现)

    本文实例讲述了Android监听手机电话状态与发送邮件通知来电号码的方法.分享给大家供大家参考,具体如下: 在android中可以用PhoneStateListener来聆听手机电话状态(比如待机.通话中.响铃等).本例是通过它来监听手机电话状态,当手机来电时,通过邮件将来电号码发送到用户邮箱的例子.具体程序如下: import android.app.Activity; import android.content.Intent; import android.os.Bundle; impor

  • Android实现监听电话呼叫状态的方法

    本文实例讲述了Android实现监听电话呼叫状态的方法.分享给大家供大家参考.具体如下: 在清单文件AndroidManifest.xml中添加权限: 复制代码 代码如下: <uses-permission android:name="android.permission.READ_PHONE_STATE"/> java代码如下: // 取得电话服务 TelephonyManager telManager = (TelephonyManager) getSystemServ

  • android获取监听SD Card状态的方法

    本文实例讲述了android获取监听SD Card状态的方法.分享给大家供大家参考.具体分析如下: 1. 注册StorageEventListener来监听SD卡状态即onStorageStateChanged()方法,当sd卡状态改变时,调用该方法. 复制代码 代码如下: public void onStorageStateChanged(String path,String oldState,String newState){ if (newState.equals(Environment.

  • android获取及监听手机网络状态

    目录 一.获取当前手机联网方式是WiFi还是手机4G数据 二.监听手机网络变化 总结 一.获取当前手机联网方式是WiFi还是手机4G数据 private boolean getNetworkType(){ ConnectivityManager mConnectivity = (ConnectivityManager)getSystemService(Context.CONNECTIVITY_SERVICE); //检查网络链接 NetworkInfo info = mConnectivity.

  • React事件监听和State状态修改方式

    目录 React事件监听和State状态修改 React事件监听相关例子 React事件监听和State状态修改 on*函数可直接使用,但只能使用在html标签上,自己创建的组件标签不可 组件可以增加state状态,修改用setState方法,修改会有缓存,只能保证在渲染到页面之前会改好.所以setState后立马去使用值可能是不对的 狗子组件1:点击会打印bark和run class Dog extends Component {   bark () {     console.log('ba

随机推荐