利用RabbitMQ实现RPC（python）

huangfen2002

　　#!/usr/bin/python
　　import pika
　　import libvirt
　　import psutil
　　import json
　　import socket
　　import os
　　import sys
　　from xml.dom import minidom
　　#配置RabbitMQ地址
　　RabbitMQServer=x.x.x.x
　　#连接libvirt，libvirt是一个虚拟机、容器管理程序。
　　def get_conn():
　　conn = libvirt.open("qemu:///system")
　　if conn == None:
　　print '--Failed to open connection to   QEMU/KVM--'
　　sys.exit(2)
　　else:
　　return conn
　　#获取虚拟机数量
　　def getVMcount():
　　conn = get_conn()
　　domainIDs = conn.listDomainsID()
　　return len(domainIDs)
　　#获取分配给所有虚拟机的内存之和
　　def getMemoryused():
　　conn = get_conn()
　　domainIDs = conn.listDomainsID()
　　used_mem = 0
　　for>in domainIDs:
　　dom = conn.lookupByID(id)
　　used_mem += dom.maxMemory()/(1024*1024)
　　return used_mem
　　#获取分配给所有虚拟机的vcpu之和
　　def getCPUused():
　　conn = get_conn()
　　domainIDs = conn.listDomainsID()
　　used_cpu = 0
　　for>in domainIDs:
　　dom = conn.lookupByID(id)
　　used_cpu += dom.maxVcpus()
　　return used_cpu
　　#获取所有虚拟机磁盘文件大小之和
　　def getDiskused():
　　conn = get_conn()
　　domainIDs = conn.listDomainsID()
　　diskused = 0
　　for>in domainIDs:
　　dom = conn.lookupByID(id)
　　xml = dom.XMLDesc(0)
　　doc = minidom.parseString(xml)
　　disks = doc.getElementsByTagName('disk')
　　for disk in disks:
　　if disk.getAttribute('device') == 'disk':
　　diskfile = disk.getElementsByTagName('source')[0].getAttribute('file')
　　diskused += dom.blockInfo(diskfile,0)[0]/(1024**3)
　　return diskused
　　#使agent.py进入守护进程模式
　　def daemonize(stdin='/dev/null',stdout='/dev/null',stderr='/dev/null'):
　　try:
　　pid = os.fork()
　　if pid > 0:
　　sys.exit(0)
　　except OSError,e:
　　sys.stderr.write("fork #1 failed: (%d) %s\n" % (e.errno,e.strerror))
　　sys.exit(1)
　　os.chdir("/")
　　os.umask(0)
　　os.setsid()
　　try:
　　pid = os.fork()
　　if pid > 0:
　　sys.exit(0)
　　except OSError,e:
　　sys.stderr.write("fork #2 failed: (%d) %s\n" % (e.errno,e.strerror))
　　sys.exit(1)
　　for f in sys.stdout,sys.stderr,: f.flush()
　　si = file(stdin,'r')
　　so = file(stdout,'a+',0)
　　se = file(stderr,'a+',0)
　　os.dup2(si.fileno(),sys.stdin.fileno())
　　os.dup2(so.fileno(),sys.stdout.fileno())
　　os.dup2(se.fileno(),sys.stderr.fileno())
　　daemonize('/dev/null','/root/kvm/agent.log','/root/kvm/agent.log')
　　#连接RabbitMQ
　　connection = pika.BlockingConnection(pika.ConnectionParameters(host= RabbitMQServer))
　　channel = connection.channel()
　　channel.exchange_declare(exchange='kvm',type='fanout')
　　result = channel.queue_declare(exclusive=True)
　　queue_name = result.method.queue
　　channel.queue_bind(exchange='kvm',queue=queue_name)
　　def on_request(ch,method,props,body):
　　sys.stdout.write(body+'\n')
　　sys.stdout.flush()
　　mem_total = psutil.virtual_memory()[0]/(1024*1024*1024)
　　cpu_total = psutil.cpu_count()
　　statvfs = os.statvfs('/datapool')
　　disk_total = (statvfs.f_frsize * statvfs.f_blocks)/(1024**3)
　　mem_unused = mem_total - getMemoryused()
　　cpu_unused = cpu_total - getCPUused()
　　disk_unused = disk_total - getDiskused()
　　data = {
　　'hostname':socket.gethostname(),#宿主机名
　　'vm':getVMcount(),#虚拟机数量
　　'available memory':mem_unused,#可用内存
　　'available cpu':cpu_unused,#可用cpu核数
　　'available disk':disk_unused#可用磁盘空间
　　}
　　json_str = json.dumps(data)
　　ch.basic_publish(exchange='',
　　routing_key=props.reply_to,
　　properties=pika.BasicProperties(correlation_id=props.correlation_id),
　　body=json_str
　　)
　　ch.basic_ack(delivery_tag=method.delivery_tag)
　　channel.basic_qos(prefetch_count=1)
　　channel.basic_consume(on_request,queue=queue_name)
　　sys.stdout.write(" [x] Awaiting RPC requests\n")
　　sys.stdout.flush()
　　channel.start_consuming()