python脚本: 计算节点创建cgroups绑定虚拟核心,实现计算资源隔离

摘要: cgroups python 计算节点 云计算 资源隔离 虚拟核心绑定

import MySQLdb

import os

import signal

import shlex

import subprocess

from eventlet.green import subprocess as green_subprocess

from eventlet import greenthread

import logging

import string

import time

mysql_host = "10.160.0.120"

mysql_db = "nova"

mysql_user = "nova"

mysql_passwd = "87da3417bb3a42ee"

mysql_port = 3306

mysql_charset = "utf8"

cgroups_hierarchy = "/home/cgroups/cpu"

LOG = logging.getLogger(__name__)

def create_process(cmd, root_helper=None, addl_env=None):
if root_helper:
cmd = shlex.split(root_helper) + cmd
cmd = map(str, cmd)
LOG.debug(("Running command: %s"), cmd)
env = os.environ.copy()
if addl_env:
env.update(addl_env)

obj = subprocess_popen(cmd, shell=False,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env)

return obj, cmd

def execute(cmd, root_helper=None, process_input=None, addl_env=None,
check_exit_code=True, return_stderr=False):
try:
obj, cmd = create_process(cmd, root_helper=root_helper,
addl_env=addl_env)
_stdout, _stderr = (process_input and
obj.communicate(process_input) or
obj.communicate())
obj.stdin.close()
m = ("\nCommand: %(cmd)s\nExit code: %(code)s\nStdout: %(stdout)r\n"
"Stderr: %(stderr)r") % {'cmd': cmd, 'code': obj.returncode,
'stdout': _stdout, 'stderr': _stderr}
LOG.debug(m)
if obj.returncode and check_exit_code:
raise RuntimeError(m)
finally:
greenthread.sleep(0)

return return_stderr and (_stdout, _stderr) or _stdout

def _subprocess_setup():
signal.signal(signal.SIGPIPE, signal.SIG_DFL)

def subprocess_popen(args, stdin=None, stdout=None, stderr=None, shell=False,
env=None):
return green_subprocess.Popen(args, shell=shell, stdin=stdin, stdout=stdout,
stderr=stderr, preexec_fn=_subprocess_setup,
close_fds=True, env=env)

def query_db_of_local_vms():
""" query controller node db for information about vms running on local host """
conn = None
try:
conn = MySQLdb.connect(host=mysql_host,
user=mysql_user,
passwd=mysql_passwd,
db=mysql_db,
port=mysql_port,
charset=mysql_charset)
except Exception as e:
#LOG.error("Fail to connect mysql .")

raise e
else:

LOG.info("Connect to mysql .")

local_compute_name = get_host()

sql = "SELECT vcpus,uuid FROM instances WHERE host='%s' AND deleted=0"%local_compute_name

vms = {}

try:
cursor = conn.cursor()
cursor.execute(sql)
result = cursor.fetchall()
cursor.close()
conn.close()

for item in result:
vms.update({item[1]:item[0]})
except Exception as ex:
#LOG.error("Exception happens while querying mysql.")

raise ex
else:
return vms

def get_host():
return os.environ['HOSTNAME']

def get_ip():
""" parse /etc/hosts to get local ip"""
df = open("/etc/hosts")
hosts = df.readlines()
hostname = get_host()
host_rec = [line for line in hosts if hostname in line and "#" not in line][0]
return host_rec.split()[0].strip()

def check_cglib():
""" check required cglibs"""
check_pkgs = "rpm -qa"
cglib_kw = "libcgroup"
cmd = check_pkgs.split()
pkgs = execute(cmd,root_helper=None)

cglibs = [pkg for pkg in pkgs.split("\n") if cglib_kw in pkg]

if len(cglibs)==0:
print "libcgroup-x.xx-x.xx.x86_64 not installed.Exit."
exit(1)

def init_cgroups():

"""ensure cgrouplib installed"""
check_cglib()

""" create cgroups base architecture """
"""clear old cgroups settings mounted on cgroups_hierarchy """
cg_clear()

"""create base cgroups hierarchy"""
create_hierarchy = ("mkdir -p %s"%cgroups_hierarchy).split()
execute(create_hierarchy,root_helper=None)

"""mount target subsystem,that's cpuset,to established hierarchy."""
mount_cg = ("mount -t cgroup -o cpuset cpuset2015  %s"%cgroups_hierarchy).split()
execute(mount_cg,root_helper=None)

def cpuinfo():
""" get cpu counts and memory nodes """
cpuinfo = "lscpu"
cmd = cpuinfo.split()
retv = execute(cmd,root_helper=None)
memNodes = retv.count("NUMA node")-1
cpus = string.atoi([line for line in retv.split("\n") if "CPU(s)" in line][0].split(":")[1].strip())
return {"cpu_count":cpus,"memNodes":memNodes}

def assign_cores():
""" cal out schemes of core binding for instances running on local host"""
vmDetails = query_db_of_local_vms()
cpuinfos = cpuinfo()
def assign(require,all):
tmp = {}
i=0
for k,v in require.iteritems():
cores = [p%all for p in xrange(i,i+v)]
i=i+v
tmp.update({k:{"cores":cores,"memNodes":"0-%d"%(cpuinfos["memNodes"]-1)}})
return tmp
vmDetails = assign(vmDetails,cpuinfos["cpu_count"])
return vmDetails

def get_pids(vmDetails):
query_pids = "ps aux"
cmd = query_pids.split()
retv = execute(cmd,root_helper=None)
qemu_kvm_processes =[p for p in retv.split("\n") if "qemu-kvm" in p and len(p)>100]
tmp = {}
for vmDetail in vmDetails:
tmp.update({vmDetail:vmDetails[vmDetail]})
for qkp in qemu_kvm_processes:
if vmDetail not in qkp:
continue
else:
pid = string.atoi(qkp.split()[1])
tmp[vmDetail]["pid"] = pid
return tmp

def create_cgs_for_instances(vmDetails):
for item in vmDetails:
detail = vmDetails[item]
if "pid" not in detail or "cores" not in detail or "memNodes" not in detail:
print "Instance %s invalid"%item
continue
else:
""" create private group for each instance"""
instance_cg = "%s/%s"%(cgroups_hierarchy,item)
create_cg_for_instance = ("mkdir -p %s"%instance_cg).split()
execute(create_cg_for_instance,root_helper=None)
setcpu_txt = "%s"%detail["cores"][0]
if len(detail["cores"]) > 1:
for core_num in detail["cores"][1:]:
setcpu_txt = "%s,%s"%(setcpu_txt,core_num)

setmem_txt = detail["memNodes"]

fd_setcpu = open("%s/cpuset.cpus"%instance_cg,"w")
fd_setcpu.write(setcpu_txt)
fd_setcpu.close()
fd_setmem = open("%s/cpuset.mems"%instance_cg,"w")
fd_setmem.write(setmem_txt)
fd_setmem.close()

print detail["pid"]
fd_task = open("%s/tasks"%instance_cg,"w")
fd_task.write("%s"%detail["pid"])
fd_task.close()

def cg_clear():
""" destroy existing cgroups """
cgclear = "umount -t cgroup %s"%cgroups_hierarchy
cmd = cgclear.split()
try:
execute(cmd,root_helper=None)
except Exception as e:
pass

def periodic_task():
init_cgroups()
hosted_vms = assign_cores()
hosted_vms = get_pids(hosted_vms)
create_cgs_for_instances(hosted_vms)

if __name__ == "__main__":
periodic_task()
while True:
time.sleep(3*60)
print "Loop Job."
periodic_task()

该脚本作用是:连接数据库查询出运行在当前宿主下的客户机的cpu配额,然后再根据当前宿主的CPU信息,做出虚拟核心的分配；创建cgroups 绑定虚拟核心，实现资源隔离.

该脚本后台运行:

nohup python /path/to/this/scripy  &