利用RabbitMQ实现RPC（python）
#!/usr/bin/pythonimport 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(" Awaiting RPC requests\n")
sys.stdout.flush()
channel.start_consuming()
页:
[1]