使用OpenStack all-in-one镜像，创建OpenStack Python运维开发环境。云主机的用户/密码为：“root/Abc@1234”，OpenStack的域名/账号/密码为：“demo/admin/000000”。
提示说明：Python脚本文件头建议加入“#encoding:utf-8”避免编码错误；测试脚本代码用python3命令执行与测试。
在controller节点的/root目录下创建api_image_manager.py脚本，编写Python代码对接OpenStack API，完成镜像的创建与上传。创建之前查询是否存在“同名镜像”，如果存在先删除该镜像。
（1）创建镜像：要求在OpenStack私有云平台中上传镜像cirros-0.3.4-x86_64-disk.img，名字为cirros001，disk_format为qcow2，container_format为bare。
（2）查询镜像：查询cirros001的详细信息，并以Json格式文本输出到控制台。
api_image_manager.py：
# ===========================================
# Copyright Jiangsu One Cloud Technology Development Co. LTD. All Rights Reserved.
# 版权： 江苏一道云科技发展有限公司，版权所有！
# ===========================================
#encoding:utf-8
import requests,json,time
def get_auth_token(controller_ip,domain,name,password):
try:
url = f"http://{controller_ip}:5000/v3/auth/tokens"
body = {
"auth": {
"identity": {
"methods": ['password'],
"password": {
"user": {
"domain": {"name": domain},
"name": name,
"password": password,
}
}
},
"scope": {
"project": {
"domain": {"name": domain},
"name": name
}
}
}
}
headers = {
"Content-Type": "application/json"
}
token = requests.post(url,headers=headers,data=json.dumps(body)).headers['X-Subject-Token']
headers = {
"X-Auth-Token": token
}
print(f"token值为:{token}")
return headers
except Exception as e:
print(f"token获取失败,{e}")
class image_manager:
def __init__(self,handers:dict,resUrl):
self.headers = handers
self.resUrl = resUrl
def create_image(self,image_name,disk_format,container_format):
body = {
"name": image_name,
"disk_format": disk_format,
"container_format": container_format,
}
req = requests.post(self.resUrl,headers=self.headers,data=json.dumps(body)).text
print(f"创建镜像的信息为:{req}")
return req
def get_image_id(self,name):
req = json.loads(requests.get(self.resUrl,headers=self.headers).text)
for image in req['images']:
if image['name'] == name:
return image['id']
return "NONE"
def upload_image(self,id,file_path:str):
url = self.resUrl + "/" + id + "/file"
self.headers["Content-Type"] = "application/octet-stream"
req = requests.put(url,headers=self.headers,data=open(file_path,'rb').read())
if req.status_code == 204:
print("上传镜像成功",req.status_code)
else:
print("上传镜像失败",req.status_code)
print(f"镜像上传信息:{req}")
return req
def get_image(self,id):
url = self.resUrl + "/" + id
req = json.loads(requests.get(self.resUrl,headers=self.headers).text)
print(f"获取到的镜像信息为:{req}")
return req
def delete_image(self,id):
url = self.resUrl + "/" + id
req = requests.delete(url,headers=self.headers)
print(f"删除信息:{req}")
return req
if __name__ == "__main__":
controller_ip = "10.26.16.133"
domain = "demo"
name = "admin"
password = "000000"
headers = get_auth_token(controller_ip, domain, name, password)
image_m = image_manager(headers,f"http://{controller_ip}:9292/v2/images")
#create
create_image = image_m.create_image("cirros001","qcow2","bare")
#get id
get_id = image_m.get_image_id("cirros001")
print(f"cirros001镜像ID为:{get_id}")
#upload
upload_image = image_m.upload_image(get_id,"cirros-0.3.4-x86_64-disk.img")
#get image
get_image = image_m.get_image(get_id)
with open("image_demo.json","w")as outfile:
json.dump(get_image,outfile,indent=4)
拉取镜像，命令如下：
[root@controller ~]# curl -O http://mirrors.douxuedu.com/newcloud/cirros-0.3.4-x86_64-disk.img
执行api_image_manager.py文件：
[root@controller ~]# python3 api_image_manager.py
------------------------------------------------执行结果---------------------------------------
token值为:gAAAAABjTgo7sOyTYBf0Kmrwvdostssitsw-G2mzvPSnFGyAdVjVjtQtTcVlEy-xM-hRs0Y30OcIzaWwQjHkJn16IoDbQ3XI8Ca4PBhv03ZmpGvlaNo9oVj6OxSH6mWpJf5yRDtzjj-1TbsYHg4AE9maaFAXXY9AvzRIw-8-V8P
WVLMrQD__oyA
创建镜像的信息为:{"container_format": "bare", "min_ram": 0, "updated_at": "2022-10-18T02:06:51Z", "file": "/v2/images/adcd6115-549b-43c6-af3a-2777ede10409/file", "owner": "0b6f2d0be
1d342e09edc31dc841db7a5", "id": "adcd6115-549b-43c6-af3a-2777ede10409", "size": null, "self": "/v2/images/adcd6115-549b-43c6-af3a-2777ede10409", "disk_format": "qcow2", "os_hash_alg
o": null, "schema": "/v2/schemas/image", "status": "queued", "tags": [], "visibility": "shared", "min_disk": 0, "virtual_size": null, "name": "cirros-test", "checksum": null, "creat
ed_at": "2022-10-18T02:06:51Z", "os_hidden": false, "protected": false, "os_hash_value": null}
cirros-test镜像ID为:adcd6115-549b-43c6-af3a-2777ede10409
上传镜像成功 204
镜像上传信息:<Response [204]>
获取到的镜像信息为:{'images': [{'container_format': 'bare', 'min_ram': 0, 'updated_at': '2022-10-18T02:06:52Z', 'file': '/v2/images/adcd6115-549b-43c6-af3a-2777ede10409/file', 'owne
r': '0b6f2d0be1d342e09edc31dc841db7a5', 'id': 'adcd6115-549b-43c6-af3a-2777ede10409', 'size': 13287936, 'self': '/v2/images/adcd6115-549b-43c6-af3a-2777ede10409', 'disk_format': 'qc
ow2', 'os_hash_algo': 'sha512', 'schema': '/v2/schemas/image', 'status': 'active', 'tags': [], 'visibility': 'shared', 'min_disk': 0, 'virtual_size': None, 'name': 'cirros-test', 'c
hecksum': 'ee1eca47dc88f4879d8a229cc70a07c6', 'created_at': '2022-10-18T02:06:51Z', 'os_hidden': False, 'protected': False, 'os_hash_value': '1b03ca1bc3fafe448b90583c12f367949f8b0e6
65685979d95b004e48574b953316799e23240f4f739d1b5eb4c4ca24d38fdc6f4f9d8247a2bc64db25d6bbdb2'}], 'schema': '/v2/schemas/images', 'first': '/v2/images'}
2. 基于Openstack Python SDK实现云主机创建
使用已建好的OpenStack Python运维开发环境，在/root目录下创建sdk_server_manager.py脚本，使用python-openstacksdk Python模块，完成云主机的创建和查询。创建之前查询是否存在“同名云主机”，如果存在先删除该镜像。
（1）创建1台云主机：云主机信息如下：
-
云主机名称：server001
-
镜像文件：cirros-0.3.4-x86_64-disk.img
-
云主机类型：m1.tiny
注意：网络等必要信息自己补充（需要创建名为“net”的网络）。
（2）查询云主机：查询云主机server001的详细信息，并以Json格式文本输出到控制台。
sdk_server_manager.py：
# ===========================================
# Copyright Jiangsu One Cloud Technology Development Co. LTD. All Rights Reserved.
# 版权： 江苏一道云科技发展有限公司，版权所有！
# ===========================================
#encoding:utf-8
import json,logging
import openstack
#文档地址
# https://docs.openstack.org/openstacksdk/latest/user/index.html
def create_connection(auth_url, user_domain_name, username, password):
"""
建立连接
"""
return openstack.connect(
auth_url=auth_url,
user_domain_name=user_domain_name,
username=username,
password=password,
)
#user Manager
# 参见文档
# https://docs.openstack.org/openstacksdk/latest/user/guides/identity.html
#openstack.connection.Connection
#云主机管理
class server_manager:
def __init__(self, connect):
self.connect = connect
def list_servers(self):
"""
查询所有云主机.
"""
#to json
items = self.connect.compute.servers()
server_jsons = {}
for server in items:
server_jsons[server['name']] = server
# return ""
return items# json.dumps(server_jsons,indent=2,skipkeys=True)
def create_server(self, server_name, image_name, flavor_name,networ_name):
image = self.connect.compute.find_image(image_name)
flavor = self.connect.compute.find_flavor(flavor_name)
network = self.connect.network.find_network(networ_name)
server = self.connect.compute.create_server(
name=server_name, image_id=image.id, flavor_id=flavor.id,
networks=[{"uuid": network.id}])
result = self.connect.compute.wait_for_server(server)
return result#json.dumps(result,indent=2,skipkeys=True)
def delete_server(self, server_name):
"""
删除云主机
"""
server = self.connect.compute.find_server(server_name)
result = self.connect.compute.delete_server(server)
return json.dumps(result, indent=2, skipkeys=True)
def get_server(self, server_name):
"""
获取云主机
"""
server = self.connect.compute.find_server(server_name)
if server:
return json.dumps(server, indent=2, skipkeys=True)
else:
return None
class image_manager:
def __init__(self, connect):
self.connect = connect
def list_images(self):
"""
查询所有镜像
"""
#to json
items = self.connect.compute.images()
images_jsons = {}
for image in items:
images_jsons[image['name']] = image
return json.dumps(images_jsons,indent=2)
def get_image(self, image_name:str):
"""
查询镜像
"""
#to json
image = self.connect.compute.find_image(image_name)
return json.dumps(image,indent=2)
class flavor_manager:
def __init__(self, connect):
self.connect = connect
def list_flavors(self):
"""
查询所有云主机类型
"""
#to json
items = self.connect.compute.flavors()
flavors_jsons = {}
for flavor in items:
flavors_jsons[flavor['name']] = flavor
return json.dumps(flavors_jsons,indent=2)
def get_flavor(self, flavor_name:str):
"""
根据名称获取云主机类.
"""
#to json
flavor = self.connect.compute.find_flavor(flavor_name)
return json.dumps(flavor,indent=2)
class network_manager:
def __init__(self, connect):
self.connect = connect
def list_networks(self):
"""
查询所有网络.
"""
#to json
items = self.connect.network.networks()
items_jsons = {}
for network in items:
items_jsons[network['name']] = network
return json.dumps(items_jsons,indent=2)
def get_network(self, network_name:str):
"""
跟名称查询网络.
"""
#to json
flavor = self.connect.compute.find_network(network_name)
return json.dumps(flavor,indent=2)
if __name__ == '__main__':
# Initialize connection(通过配置文件）
# controller_ip = "10.24.2.22"
controller_ip = "controller"
auth_url = "http://controller:5000/v3/"
username = "admin"
password = "000000"
user_domain_name = 'demo'
conn = create_connection(auth_url, user_domain_name, username, password)
sdk_m = server_manager(conn)
server = sdk_m.get_server("server001")
if server:
result = sdk_m.delete_server("server001")
print("servers:", result)
#2 创建云主机
print("creat server--------")
servers = sdk_m.create_server("server001","cirros001","m1.tiny","net")
print("servers:", servers)
#6 查询云主机
server_info = sdk_m.get_server("server001")
print(server_info)
创建网络，命令如下：
[root@controller ~]# source /etc/keystone/admin-openrc.sh
[root@controller ~]# openstack network create --provider-network-type vlan --provider-physical-network provider --provider-segment 10 --project admin net
[root@controller ~]# NET="111.111.10.0/24"
[root@controller ~]# ID=$(openstack network list --project admin |grep -v ID |grep net |awk -F "| " {'print $2'})
[root@controller ~]# openstack subnet create --project admin --subnet-range $NET --dhcp --network $ID ext-subnet
执行sdk_server_manager.py文件：
[root@controller ~]# python3 sdk_server_manager.py
-----------------------------------------执行结果-----------------------------
#创建云主机的信息为：{"server": {"security_groups": [{"name": "default"}], "OS-DCF:diskConfig": "MANUAL", "id": "f295eb8d-52a0-4e21-ba86-6
4ba8c3e6359", "links": [{"href": "http://192.168.106.10:8774/v2.1/servers/f295eb8d-52a0-4e21-ba86-64ba8c3e6359", "rel": "self"}, {"href":
"http://192.168.106.10:8774/servers/f295eb8d-52a0-4e21-ba86-64ba8c3e6359", "rel": "bookmark"}], "adminPass": "oTs5jHFwkbWy"}}
#获取到的云主机信息为:{'server': {'OS-EXT-STS:task_state': None, 'addresses': {'net': [{'OS-EXT-IPS-MAC:mac_addr': 'fa:16:3e:ce:f7:82', 'v
ersion': 4, 'addr': '192.168.10.138', 'OS-EXT-IPS:type': 'fixed'}]}, 'links': [{'href': 'http://192.168.106.10:8774/v2.1/servers/f295eb8d
-52a0-4e21-ba86-64ba8c3e6359', 'rel': 'self'}, {'href': 'http://192.168.106.10:8774/servers/f295eb8d-52a0-4e21-ba86-64ba8c3e6359', 'rel':
'bookmark'}], 'image': {'id': 'aa5005f1-54e3-4636-9b6d-83b84e0765dd', 'links': [{'href': 'http://192.168.106.10:8774/images/aa5005f1-54e
3-4636-9b6d-83b84e0765dd', 'rel': 'bookmark'}]}, 'OS-EXT-STS:vm_state': 'active', 'OS-EXT-SRV-ATTR:instance_name': 'instance-00000097', '
OS-SRV-USG:launched_at': '2022-07-30T16:02:27.000000', 'flavor': {'id': '10', 'links': [{'href': 'http://192.168.106.10:8774/flavors/10',
'rel': 'bookmark'}]}, 'id': 'f295eb8d-52a0-4e21-ba86-64ba8c3e6359', 'security_groups': [{'name': 'default'}], 'user_id': '7528400cda4243
af96196e790cc9cf33', 'OS-DCF:diskConfig': 'MANUAL', 'accessIPv4': '', 'accessIPv6': '', 'progress': 0, 'OS-EXT-STS:power_state': 1, 'OS-E
XT-AZ:availability_zone': 'nova', 'config_drive': '', 'status': 'ACTIVE', 'updated': '2022-07-30T16:02:27Z', 'hostId': 'bbbe102b2eaf03e47
ef7d63ff01ee9c176f4d54c744e7d0693e8135a', 'OS-EXT-SRV-ATTR:host': 'compute1', 'OS-SRV-USG:terminated_at': None, 'key_name': None, 'OS-EXT
-SRV-ATTR:hypervisor_hostname': 'compute1', 'name': 'server001', 'created': '2022-07-30T16:02:18Z', 'tenant_id': 'f94153beea9d490382649f0
96af3d512', 'os-extended-volumes:volumes_attached': [], 'metadata': {}}}
3. 基于Python arg云主机类型管理的命令行工具开发
使用已建好的OpenStack Python运维开发环境，在/root目录下创建flavor_manager.py脚本，完成云主机类型的管理，flavor_manager.py程序支持命令行参数执行。
提示说明：Python标准库argparse模块，可以提供命令行参数的解析。
要求如下：
（1）程序支持根据命令行参数，创建1个多云主机类型。返回response。
参考运行实例：
python3 flavor_manager.py create -n flavor_small -m 1024 -v 1 -d 10 -id 100000
-
位置参数“create”，表示创建；
-
参数“-n”支持指定flavor名称，数据类型为字符串类型；
-
参数“-m”支持指定内存大小，数据类型为int，单位M；
-
参数“-v”支持指定虚拟cpu个数，数据类型为int；
-
参数“-d”支持磁盘大小，内存大小类型为int，单位G；
-
参数“-id”支持指定ID，类型为字符串。
（2）程序支持查询目前admin账号下所有的云主机类型。
参考执行实例如下：
python3 flavor_manager.py getall
-
位置参数“getall”，表示查询所有云主机类型；
-
查询结果，以Json格式输出到控制台。
（3）支持查询给定具体名称的云主机类型查询。控制台以Json格式输出创建结果。
参考执行实例如下：
python3 flavor_manager.py get -id 100000
-
位置参数“get”，表示查询1个云主机类型；
-
参数“-id”支持指定ID查询，类型为string。
（4）支持删除指定的ID云主机类型。
参考执行实例如下：
python3 flavor_manager.py delete -id 100001
-
位置参数“delete”，表示删除一个云主机类型；
-
参数“-id”支持指定ID查询，返回response，控制台输出response。
api_flavor_manager.py：
# ===========================================
# Copyright Jiangsu One Cloud Technology Development Co. LTD. All Rights Reserved.
# 版权： 江苏一道云科技发展有限公司，版权所有！
# ===========================================
#encoding:utf-8
# Copyright 2021~2022 The Cloud Computing support Teams of ChinaSkills.
import requests,json,time
import logging
#-----------logger-----------
#get logger
logger = logging.getLogger(__name__)
# level
logger.setLevel(logging.DEBUG)
# format
format = logging.Formatter('%(asctime)s %(message)s')
# to console
stream_handler = logging.StreamHandler()
stream_handler .setFormatter(format)
logger.addHandler(stream_handler )
#-----------logger-----------
def get_auth_token(controller_ip, domain, user, password):
try:
url = f"http://{controller_ip}:5000/v3/auth/tokens"
body = {
"auth": {
"identity": {
"methods": [
"password"
],
"password": {
"user": {
"domain": {
"name": domain
},
"name": user,
"password": password
}
}
},
"scope": {
"project": {
"domain": {
"name": domain
},
"name": user
}
}
}
}
headers = {
"Content-Type": "application/json",
}
print(body)
Token = requests.post(url, data=json.dumps(body), headers=headers).headers['X-Subject-Token']
headers = {
"X-Auth-Token": Token
}
logger.debug(f"获取Token值：{str(Token)}")
return headers
except Exception as e:
logger.error(f"获取Token值失败，请检查访问云主机控制节点IP是否正确？输出错误信息如下：{str(e)}")
exit(0)
class flavor_manager:
def __init__(self,handers:dict,resUrl:str):
self.headers=handers
self.resUrl=resUrl
#创建flavor类型
def create_flavor(self,flavor_name:str,ram,vcpus,disk,id):
self.headers['Content-Type']="application/json"
body={
"flavor":{
"name":flavor_name,
"ram":ram,
"vcpus":vcpus,
"disk":disk,
"id":id,
}
}
logger.debug(f"创建flavor请求body:{str(body)}")
status_code = requests.post(self.resUrl, data=json.dumps(body), headers=self.headers).text
logger.debug(f"返回状态:{str(status_code)}")
return status_code
#获取all flavors
def get_flavors(self):
result = json.loads(requests.get(self.resUrl,headers=self.headers).text)
logger.debug(f"返回信息:{str(result)}")
return result
# 获取flavor_id
def get_flavor(self, id:str):
api_url = self.resUrl + "/"+id
result = json.loads(requests.get(api_url, headers=self.headers).text)
logger.debug(f"返回信息:{str(result)}")
return result
def delete_flavor(self, id:str):
api_url = self.resUrl + "/"+id
response = requests.delete(api_url, headers=self.headers)
#Normal response codes: 202 without return text
if response.status_code == 202:
return {"itemDeletedSuccess": response.status_code}
result = json.loads(response.text)
logger.debug(f"返回信息:{str(result)}")
return result
#http://192.168.200.226:8774/v2.1/ get apis version infomation.
def update_flavor_desc(self, id: str, desc:str):
# 特别注意：This API is available starting with microversion 2.55.
self.headers['X-OpenStack-Nova-API-Version'] = "2.55"
self.headers['Content-Type'] = "application/json"
body = {
"flavor": {
"description": desc
}
}
api_url = self.resUrl + "/" + id
response = requests.put(api_url, data=json.dumps(body), headers=self.headers)
# Normal response codes: 202 without return text
if response.status_code == 202:
return {"itemUpdateSuccess": response.status_code}
result = json.loads(response.text)
logger.debug(f"返回信息:{str(result)}")
return result
if __name__ == '__main__':
controller_ip = "controller"
domain = "demo"
user = "admin"
password = "000000"
headers = get_auth_token(controller_ip, domain, user, password)
flavor_m = flavor_manager(headers, f"http://{controller_ip}:8774/v2.1/flavors")
#1 查所有
flavors = flavor_m.get_flavors()
print("查询所有flavors:", flavors)
flavor_manager.py：
# ===========================================
# Copyright Jiangsu One Cloud Technology Development Co. LTD. All Rights Reserved.
# 版权： 江苏一道云科技发展有限公司，版权所有！
# ===========================================
#encoding:utf-8
import argparse
import api_flavor_manager
# 1. openstack allinone （controller ) credentials
# host ip address
# controller_ip = "10.24.2.22"
controller_ip = "controller"
# domain name
domain = "demo"
# user name
user = "admin"
# user password
password = "000000"
headers = api_flavor_manager.get_auth_token(controller_ip,domain,user,password)
print("headers:", headers)
#. get token
flavor_m = api_flavor_manager.flavor_manager(headers, "http://controller:8774/v2.1/flavors")
def define_args(parser):
"""
定义程序支持的args
:return:
"""
# parser = argparse.ArgumentParser()
#增加控制命令(postion 位置参数，必须）
parser.add_argument('command',
help='Resource command name',
type=str)
# parser.add_argument('delete',
# help='delete a resource',
# type=str)
#可选参数（可有可无）
parser.add_argument('-n', '--name', # 可选参数,删除的名称
help='The Name of the resource', # 输入-h展示
type=str)
parser.add_argument('-o', '--output', # 可选参数,删除的名称
help='The output file path ', # 输入-h展示
type=str)
parser.add_argument('-m', '--memory', # 可选参数,删除的名称
help='The Name of the resource', # 输入-h展示
type=str)
parser.add_argument('-v', '--vcpu', # 可选参数,删除的名称
help='The Name of the resource', # 输入-h展示
type=str)
parser.add_argument('-d', '--disk', # 可选参数,删除的名称
help='The Name of the resource', # 输入-h展示
type=str)
parser.add_argument('-id', '--id', # 可选参数,删除的名称
help='The Name of the resource', # 输入-h展示
type=str)
def parse_args(parser):
args = parser.parse_args()
if args.command:
if args.command == "create":
print("create some thing")
create_flavor(args)
elif args.command == "getall":
print("getall some thing")
getall_flavor(args)
elif args.command == "get":
print("get some thing")
get_flavor(args)
elif args.command == "delete":
print("delete some thing")
delete_flavor(args)
else:
print("Note support command name!")
def create_flavor(args):
print('Provided command value is %r.' % args.command)
print('Provided name value is %r.' % args.name)
print('Provided memory value is %r.' % args.memory)
print('Provided vcpu value is %r.' % args.vcpu)
print('Provided disk value is %r.' % args.disk)
print('Provided id value is %r.' % args.id)
result = flavor_m.create_flavor(args.name,args.memory,args.vcpu,args.disk,args.id)
print(result)
def delete_flavor(args):
print('Provided command value is %r.' % args.command)
print('Provided id value is %r.' % args.id)
result = flavor_m.delete_flavor(args.id)
print(result)
def getall_flavor(args):
print('Provided command value is %r.' % args.command)
result = flavor_m.get_flavors()
print(result)
def get_flavor(args):
print('Provided command value is %r.' % args.command)
print('Provided id value is %r.' % args.id)
result = flavor_m.get_flavor(args.id)
print(result)
if __name__ == '__main__':
parser = argparse.ArgumentParser()
define_args(parser)
parse_args(parser)
-------------------------------------------执行结果--------------------------------
{'auth': {'identity': {'methods': ['password'], 'password': {'user': {'domain': {'name': 'demo'}, 'name': 'admin', 'password': '000000'}}}, 'scope': {'project': {'domain': {'name': 'demo'}, 'name': 'admin'}}}}
2022-10-9 17:30:56,409 获取Token值：gAAAAABjRTfQ5U6ZcSNZb3u_ipsk3GKd_Ch71mizNSDefcV3BtBzD-EsoD76Ru0TGWukQRWI5JpVnFQH4Wsb2fJYHsgtur3n-niV32c9l9SvBqG5039lXOwelSywADqpe3ziRukwV7yHeAZICgXxFt_cCz4FNv5qT-NJhRTJJ0xZxZ73PS4R9pw
headers: {'X-Auth-Token': 'gAAAAABjRTfQ5U6ZcSNZb3u_ipsk3GKd_Ch71mizNSDefcV3BtBzD-EsoD76Ru0TGWukQRWI5JpVnFQH4Wsb2fJYHsgtur3n-niV32c9l9SvBqG5039lXOwelSywADqpe3ziRukwV7yHeAZICgXxFt_cCz4FNv5qT-NJhRTJJ0xZxZ73PS4R9pw'}
create some thing
Provided command value is 'create'.
Provided name value is 'flavor_small'.
Provided memory value is '1024'.
Provided vcpu value is '1'.
Provided disk value is '10'.
Provided id value is '100000'.
2022-10-9 17:30:56,410 创建flavor请求body:{'flavor': {'name': 'flavor_small', 'ram': '1024', 'vcpus': '1', 'disk': '10', 'id': '100000'}}
2022-10-9 17:30:56,952 返回状态:{"flavor": {"links": [{"href": "http://controller:8774/v2.1/flavors/100000", "rel": "self"}, {"href": "http://controller:8774/flavors/100000", "rel": "bookmark"}], "ram": 1024, "OS-FLV-DISABLED:disabled": false, "os-flavor-access:is_public": true, "rxtx_factor": 1.0, "disk": 10, "id": "100000", "name": "flavor_small", "vcpus": 1, "swap": "", "OS-FLV-EXT-DATA:ephemeral": 0}}
{"flavor": {"links": [{"href": "http://controller:8774/v2.1/flavors/100000", "rel": "self"}, {"href": "http://controller:8774/flavors/100000", "rel": "bookmark"}], "ram": 1024, "OS-FLV-DISABLED:disabled": false, "os-flavor-access:is_public": true, "rxtx_factor": 1.0, "disk": 10, "id": "100000", "name": "flavor_small", "vcpus": 1, "swap": "", "OS-FLV-EXT-DATA:ephemeral": 0}}
4. 基于Python arg用户管理的命令行工具开发
使用已建好的OpenStack Python运维开发环境，在/root目录下创建user_manager.py脚本，完成用户管理功能开发，user_manager.py程序支持命令行带参数执行。
提示说明：Python标准库argparse模块，可以提供命令行参数的解析。
（1）程序支持根据命令行参数，创建1个用户。查询结果，以Json格式输出到控制台。
参考执行实例如下：
python3 user_manager.py create --input '{ "name": "user01", "password": "000000", "description": "description" } '
-
位置参数“create”，表示创建；
-
参数“-i 或–input”，格式为Json格式文本用户数据。
（2）支持查询给定具体名称的用户查询。
参考执行实例如下：
python3 user_manager.py get --name user01 -o user.json
-
位置参数“get”，表示查询1个用户；
-
参数“-n 或 –name”支持指定名称查询，类型为string。
-
参数“-o 或 output”支持查询该用户信息输出到文件，格式为Json格式。
（3）程序支持查询目前admin账号下所有的用户。
参考执行实例如下：
python3 user_manager.py getall -o openstack_all_user.yaml
-
位置参数“getall”，表示查询所有用户；
-
参数“-o 或–output”支持输出到文件，格式为yaml格式。
（4）支持删除指定的名称的用户。
参考执行实例如下：
python3 user_manager.py delete --name user01
-
位置参数“delete”，表示删除一个用户；返回response，通过控制台输出。
-
参数“-n或–name”支持指定名称查询，类型为string。
api_user_manager.py：
# ===========================================
# Copyright Jiangsu One Cloud Technology Development Co. LTD. All Rights Reserved.
# 版权： 江苏一道云科技发展有限公司，版权所有！
# ===========================================
# encoding:utf-8
import requests, json, time
import logging
# -----------logger-----------
# get logger
logger = logging.getLogger(__name__)
# level
logger.setLevel(logging.DEBUG)
# format
format = logging.Formatter('%(asctime)s %(message)s')
# to console
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(format)
logger.addHandler(stream_handler)
# -----------logger-----------
def get_auth_token(controller_ip, domain, user, password):
'''
:param controller_ip: openstack master ip address
:param domain: current user's domain
:param user: user name
:param password: user password
:return: keystoen auth Token for current user.
'''
try:
url = "http://controller:5000/v3/auth/tokens"
body = {
"auth": {
"identity": {
"methods": [
"password"
],
"password": {
"user": {
"domain": {
"name": domain
},
"name": user,
"password": password
}
}
},
"scope": {
"project": {
"domain": {
"name": domain
},
"name": user
}
}
}
}
headers = {
"Content-Type": "application/json",
}
print(body)
Token = requests.post(url, data=json.dumps(body), headers=headers).headers['X-Subject-Token']
headers = {
"X-Auth-Token": Token
}
logger.debug(f"获取Token值：{str(Token)}")
return headers
except Exception as e:
logger.error(f"获取Token值失败，请检查访问云主机控制节点IP是否正确？输出错误信息如下：{str(e)}")
exit(0)
# 用户管理
# https://docs.openstack.org/api-ref/identity/v3/index.html#users
class user_manager:
def __init__(self, handers: dict, resUrl: str):
self.headers = handers
self.resUrl = resUrl
# POST /v3/users Create user
def create_users(self, user_name, password: str, desc: str):
"""
create a user with name and password and description.
"""
body = {
"user": {
"name": user_name,
"password": password,
"description": desc,
}
}
status_code = requests.post(self.resUrl, data=json.dumps(body), headers=self.headers).text
logger.debug(f"返回状态:{str(status_code)}")
return status_code
# /v3/users # List all users
def get_users(self):
"""
get user
"""
status_code = requests.get(self.resUrl, headers=self.headers).text
logger.debug(f"返回状态:{str(status_code)}")
return status_code
def get_user_id(self, user_name):
"""
get user id by name.
"""
result = json.loads(requests.get(self.resUrl, headers=self.headers).text)
user_name = user_name
for item in result['users']:
if item['name'] == user_name:
return item['id']
return "NONE"
def get_user(self, id: str):
"""
get a flavor by id.
"""
api_url = self.resUrl + "/" + id
result = json.loads(requests.get(api_url, headers=self.headers).text)
logger.debug(f"返回信息:{str(result)}")
return result
def delete_user(self, name: str):
"""
delete a user by id.
"""
id = self.get_user_id(name)
api_url = self.resUrl + "/" + id
response = requests.delete(api_url, headers=self.headers)
# 204 - No Content The server has fulfilled the request.
if response.status_code == 204:
return {"User itemDeletedSuccess": response.status_code}
result = json.loads(response.text)
logger.debug(f"返回信息:{str(result)}")
return result
# http://192.168.200.226:8774/v2.1/ get apis version infomation.
def update_User_password(self, id: str, original_password: str, new_password: str):
"""
update a flavor desc by id.
"""
self.headers['Content-Type'] = "application/json"
body = {
"user": {
"password": new_password,
"original_password": original_password
}
}
api_url = self.resUrl + "/" + id + "/password"
response = requests.post(api_url, data=json.dumps(body), headers=self.headers)
# Normal response codes: 204 without return text
if response.status_code == 204:
return {"item Update Password Success": response.status_code}
result = json.loads(response.text)
logger.debug(f"返回信息:{str(result)}")
return result
if __name__ == '__main__':
# 1. openstack allinone （controller ) credentials
# host ip address
# controller_ip = "10.24.2.22"
controller_ip = "controller"
# controller_ip = "10.24.2.22"
# domain name
domain = "demo"
# user name
user = "admin"
# user password
password = "000000"
headers = get_auth_token(controller_ip, domain, user, password)
print("headers:", headers)
# get all user
user_m = user_manager(headers, "http://controller:5000/v3/users")
# 1 查询所有
users = user_m.get_users()
print("查询所有users:", users)
user_manager.py：
# ===========================================
# Copyright Jiangsu One Cloud Technology Development Co. LTD. All Rights Reserved.
# 版权： 江苏一道云科技发展有限公司，版权所有！
# ===========================================
#encoding:utf-8
import argparse
import api_user_manager
import json
import csv
import yaml
# 1. openstack allinone （controller ) credentials
# host ip address
# controller_ip = "10.24.2.22"
controller_ip = "controller"
# controller_ip = "10.24.2.22"
# domain name
domain = "demo"
# user name
user = "admin"
# user password
password = "000000"
headers = api_user_manager.get_auth_token(controller_ip, domain, user, password)
print("headers:", headers)
# get all user
user_m = api_user_manager.user_manager(headers, "http://controller:5000/v3/users")
print("-----------begin-----------------")
def define_args(parser):
"""
定义程序支持的args
:return:
"""
# parser = argparse.ArgumentParser()
#增加控制命令(postion 位置参数，必须）
parser.add_argument('command',
help='Resource command name',
type=str)
# parser.add_argument('delete',
# help='delete a resource',
# type=str)
#可选参数（可有可无）
parser.add_argument('-n', '--name', # 可选参数,删除的名称
help='The Name of the resource', # 输入-h展示
type=str)
parser.add_argument('-i', '--input', # 可选参数,删除的名称
help='The input json format text ', # 输入-h展示
type=str)
parser.add_argument('-o', '--output', # 可选参数,删除的名称
help='The output file path ', # 输入-h展示
type=str)
def parse_args(parser):
args = parser.parse_args()
if args.command:
if args.command == "create":
print("create some thing")
create_user(args)
elif args.command == "getall":
print("getall some thing")
getall_users(args)
elif args.command == "get":
print("get some thing")
get_user(args)
elif args.command == "delete":
print("delete some thing")
delete_user(args)
else:
print("Note support command name!")
def create_user(args):
print('Provided command value is %r.' % args.command)
print('Provided input value is %r.' % args.input)
print('Provided output value is %r.' % args.output)
output_file = args.output
# user_name, password: str, desc: str):
user_dict = json.loads(args.input)
result = user_m.create_users(user_dict["name"],user_dict["password"],user_dict["description"])
# 写出json文件
print("--------write to json---------:", result)
print(result)
def delete_user(args):
print('Provided command value is %r.' % args.command)
print('Provided input value is %r.' % args.input)
print('Provided output value is %r.' % args.output)
result = user_m.delete_user(args.name)
print(result)
def getall_users(args):
print('Provided command value is %r.' % args.command)
print('Provided input value is %r.' % args.input)
print('Provided output value is %r.' % args.output)
print(type(args.input))
result = user_m.get_users()
output_file = args.output
# 写出json文件
print("--------result---------")
print(result)
configuration = json.loads(result)
# 写出yaml (dict)
with open(output_file, 'w') as yaml_file:
yaml.dump(configuration, yaml_file)
print(result)
def get_user(args):
print('Provided command value is %r.' % args.command)
print('Provided input value is %r.' % args.input)
print('Provided output value is %r.' % args.output)
id = user_m.get_user_id(args.name)
result = user_m.get_user(id)
output_file = args.output
# 写出json文件
with open(output_file, 'w') as jsonfile:
json.dump(result, jsonfile, indent=4)
print(result)
if __name__ == '__main__':
import sys
print(sys.argv)
parser = argparse.ArgumentParser()
define_args(parser)
parse_args(parser)
----------------------------------------------执行结果------------------------------------
[root@controller python]# python3 user_manager.py create --input '{ "name": "user01", "password": "000000", "description": "description" } '
{'auth': {'identity': {'methods': ['password'], 'password': {'user': {'domain': {'name': 'demo'}, 'name': 'admin', 'password': '000000'}}}, 'scope': {'project': {'domain': {'name': 'demo'}, 'name': 'admin'}}}}
2022-10-11 17:39:22,843 获取Token值：gAAAAABjRTnKtdV9oDS_VfNDp8qtRC_sEElsQwJGqJTST8LHtqJUahTJtf8MVDa2Nplrjwo6_18D_Hm85j99D9G1TMq7jKEPqAynBx5nGkTXggQWJ-WJdPxad_e3qsrwfeL3JOqDK3RSHEkhZ1k1EQKWl3nxgMBhycHDs_3-CA4Cyfcmi9S15pQ
headers: {'X-Auth-Token': 'gAAAAABjRTnKtdV9oDS_VfNDp8qtRC_sEElsQwJGqJTST8LHtqJUahTJtf8MVDa2Nplrjwo6_18D_Hm85j99D9G1TMq7jKEPqAynBx5nGkTXggQWJ-WJdPxad_e3qsrwfeL3JOqDK3RSHEkhZ1k1EQKWl3nxgMBhycHDs_3-CA4Cyfcmi9S15pQ'}
-----------begin-----------------
['user_manager.py', 'create', '--input', '{ "name": "user01", "password": "000000", "description": "description" } ']
create some thing
Provided command value is 'create'.
Provided input value is '{ "name": "user01", "password": "000000", "description": "description" } '.
Provided output value is None.
2022-10-11 17:39:23,137 返回状态:{"user": {"description": "description", "name": "user01", "domain_id": "default", "enabled": true, "links": {"self": "http://controller:5000/v3/users/01eebcdbcbf24bc4a5435f1dcd0949a7"}, "options": {}, "id": "01eebcdbcbf24bc4a5435f1dcd0949a7", "password_expires_at": null}}
--------write to json---------: {"user": {"description": "description", "name": "user01", "domain_id": "default", "enabled": true, "links": {"self": "http://controller:5000/v3/users/01eebcdbcbf24bc4a5435f1dcd0949a7"}, "options": {}, "id": "01eebcdbcbf24bc4a5435f1dcd0949a7", "password_expires_at": null}}
{"user": {"description": "description", "name": "user01", "domain_id": "default", "enabled": true, "links": {"self": "http://controller:5000/v3/users/01eebcdbcbf24bc4a5435f1dcd0949a7"}, "options": {}, "id": "01eebcdbcbf24bc4a5435f1dcd0949a7", "password_expires_at": null}}