前言: 注意是扩展开发,这个词是我杜撰的,大概意思是指基于 openstack 的 rest api 做的一些开发,用于辅助相关功能,而不是直接改动 openstack 内的代码,怎么修改添加 openstack 各个组件的代码不在此文章内容内。
首先,千万,千万,千万不要用 Openstack 提供的 SDK,原因如下。
一,SDK 的相关文档并不健全。
二,版本不够统一,即兼容的问题。
所以不要使用 openstack 的 SDK 而是自己查阅 openstack 的 API 文档,通过 requests 库发 http 请求要比 SDK 灵活并便捷得多的方式。但是难过的地方在于开头,这也是本章的主要内容。为了使内容更加贴近现实,我们要做的需求是 Openstack 计算节点的高可用,主题内容如下:
一:解决思路
二:查询相关 API
三:组织代码
四:总结
(一) 解决思路
其实 openstack 计算节点高可用的方案应该是不同的厂商有不同解决方案,这里不具体分析各个方案,二是针对笔者自己实现的一个方案的实现讲解。
一,怎样确定计算节点挂了
二,怎样将计算节点上的虚拟机迁移到现存可用的计算节点
三,迁移之后是否应该采取相关措施
笔者的对应的解决方案如下,
一:通过 openstack 自身机制监控,即计算节点不可用时,不会上报状态,控制节点会将该计算节点的状态 state 设置为 down,为 down 的就判定该计算节点不可用。
二:通过 evacuate API 将不可用的计算节点上的虚拟机迁移到可用的计算节点,共享存储肯定是必须的。
三:通过将该不可用计算节点 status 设置为 disable,以达到隔离效果,即该计算节点即使又好了,也不会将新的虚拟机调度到该计算节点,除非人为干预。
注:这里没有考虑网络环境。
(二) 查询相关 API
一:所有计算节点可用状态:/os-services
二:查询指定计算节点上的虚拟机:/servers/detail
三:evacuate 实例:/servers/<server-id>/action
四:disable 阶段节点:/os-services/disable
主要就是上面这些 API 了。
(三) 组织代码
1. 首先是认证获取 token,获取各个 endpoint。
2. 统一封装 HTTP 请求过程。
3. 然后为每个对应的 API 封装一个函数。
4. 将整个需求的逻辑独立与封装的函数及类。
相对应的代码如下
1. 这里新建一个 baseInfo 的类作为基类,用于获取 token 及相应的 endpoint
- class baseInfo(object):
- """init the info of all compute, and get the token for access the api"""
- def __init__(self):
- confFile = sys.argv[1]
- headers = {}
- headers["Content-Type"] = "application/json"
- self.cf = ConfigParser()
- self.cf.read(confFile)
- self.conf = self.getConf()
- self.headers = headers
- self.catalog, self.token = self.getToken()
- self.url = [url for url in self.catalog if url["name"] == "nova"]
- self.url = self.url[0]["endpoints"][0]["publicURL"]
- def getConf(self):
- try:
- conf = {
- "url": self.cf.get("ser","OS_AUTH_URL"),
- "uname" : self.cf.get("ser","OS_USERNAME"),
- "passwd" : self.cf.get("ser","OS_PASSWORD"),
- "tname" : self.cf.get("ser","OS_TENANT_NAME"),
- "interval" : self.cf.get("ser","INTERVAL")}
- except Exception as e:
- logging.critical("加载配置文件失败")
- logging.critical(e)
- sys.exit(1)
- return conf
- def getToken(self):
- headers = self.headers
- url = self.conf["url"] + "/tokens"
- data = '{"auth": {"tenantName": "%s", "passwordCredentials": {"username": "%s", "password": "%s"}}}'
- data = data % (self.conf["tname"], self.conf["uname"], self.conf["passwd"])
- try:
- logging.debug("开始获取Token")
- ret = requests.post(url, data=data, headers=headers)
- logging.debug("request url:%s" % ret.url)
- ret = ret.json()
- except Exception as e:
- msg = "获取Token失败 data:%s headers:%s" % (data, headers)
- logging.critical(msg)
- logging.critical(e)
- catalog = ret["access"]["serviceCatalog"]
- token = ret["access"]["token"]["id"]
- return catalog, token
2. 将每个交互 API 的 HTTP 请求独立出来,所谓 DRY 吧,不要重复你自己,这样的好处自然是不用重复写代码,再者就是在一个统一的地方对所有调用了的,统一包装或修改。
- def getResp(self, suffix, method, data=None, headers=None, params=None, isjson=True):
- """return the result of requests"""
- url = self.url + suffix
- if headers == None:
- headers = self.headers.copy()
- headers["X-Auth-Token"] = self.token
- req = getattr(requests, method)
- try:
- ret = req(url, data=data, headers=headers, params=params, verify=False)
- logging.debug("request url:%s" % ret.url)
- except Exception as e:
- msg = "%s访问%s失败 data:%s headers:%s" % (method, suffix, data, headers)
- logging.critical(msg)
- logging.critical(e)
- sys.exit(1)
- if ret.status_code == 401:
- logging.warning("Token 过期,重新获取Token")
- self.catalog, self.token = self.getToken()
- headers["X-Auth-Token"] = self.token
- logging.debug("request headers:%s" % ret.request.headers)
- ret = req(url, data=data, headers=headers)
- if isjson:
- ret = ret.json()
- return ret
3. 封装每个 API 作为函数,函数是第一公民嘛。
- ...
- def chkNode(self):
- """get the compute list service down"""
- suffix = "/os-services"
- ret = self.getResp(suffix, "get")
- ret = ret["services"]
- cmpAll = [host["host"] for host in ret if host["binary"] == "nova-compute"]
- cmpDown = [host["host"] for host in ret if host["state"] != "up" and host["binary"] == "nova-compute"]
- return cmpDown, cmpAll
- def chkSerFromNode(self):
- """get the server list from failed node"""
- suffix = "/servers/detail"
- params = {"all_tenants":1}
- ret = self.getResp(suffix, "get", params=params)
- ret = ret["servers"]
- serDown = [ser["id"] for ser in ret if ser["OS-EXT-SRV-ATTR:host"] in self.cmpDown]
- self.serDown.extend(serDown)
- def evacuate(self, serID):
- """evacuate the server"""
- suffix = "/servers/%s/action" % serID
- data = '{"evacuate": {"onSharedStorage": "True"}}'
- ret = self.getResp(suffix, "post", data=data, isjson=False)
- return ret.ok
- ...
4. 逻辑独立出来,不要做太多事情。
- def main():
- if len(sys.argv) > 1 and os.path.isfile(sys.argv[1]):
- ch = check()
- fen = fence()
- recov = recover()
- while True:
- interval = ch.conf["interval"]
- try:
- interval = int(interval)
- except Exception as e:
- msg = "时间间隔设置有误 interval:%s" % interval
- logging.critical(msg)
- logging.critical(e)
- sys.exit(1)
- cmd = "pcs status|grep 'Current DC'|grep `hostname`"
- p = sp.Popen(cmd, shell=True, stdout=DEVNULL, stderr=sp.STDOUT)
- vip = p.wait()
- if not vip:
- ch.check()
- logging.info("失败的计算节点%s" %ch.cmpDown)
- fen.fence(ch.cmpDown)
- logging.info("失败的计算节点下面的server %s" %ch.serDown)
- recov.recover(ch.serDown)
- time.sleep(interval)
- else:
- print "配置文件不存在"
(四) 总结
不要用 SDK,rest API 的文档足够健全。
至于完整项目可参考我的 Github。
相关链接:
Openstack API:
来源: http://www.bubuko.com/infodetail-1989222.html