案例描述

本案例讲述了OpenStack利用Python调用API以及SDK实现私有云自动化运维，基于Python args模块，封装OpenStack API实现开发OpenStack命令行管理工具

规划节点

节点规划见表1-1。

表1-1 节点规划

IP 主机名 节点
192.168.200.21 controller controller

基础准备

使用已经部署好OpenStack平台的openstack-allinone镜像。

下载本案例所需用到的附件以及代码文件：

1
openstack-python.tar.gz

案例实施

基于OpenStack Restful API实现镜像上传

使用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：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
#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)

把所需镜像上传，如下：

1
2
[root@controller ~]# ls
cirros-0.3.4-x86_64-disk.img

执行api_image_manager.py文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
[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'}

基于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：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
#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)

创建网络，命令如下：

1
2
3
4
5
[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文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
[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': {}}}

基于Python arg云主机类型管理的命令行工具开发

使用已建好的OpenStack Python运维开发环境，在/root目录下创建flavor_manager.py脚本，完成云主机类型的管理，flavor_manager.py程序支持命令行参数执行。

提示说明：Python标准库argparse模块，可以提供命令行参数的解析。

要求如下：

（1）程序支持根据命令行参数，创建1个多云主机类型。返回response。

参考运行实例：

1
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账号下所有的云主机类型。

参考执行实例如下：

1
python3 flavor_manager.py getall
  • 位置参数“getall”，表示查询所有云主机类型；
  • 查询结果，以Json格式输出到控制台。

（3）支持查询给定具体名称的云主机类型查询。控制台以Json格式输出创建结果。

参考执行实例如下：

1
python3 flavor_manager.py get -id 100000
  • 位置参数“get”，表示查询1个云主机类型；
  • 参数“-id”支持指定ID查询，类型为string。

（4）支持删除指定的ID云主机类型。

参考执行实例如下：

1
python3 flavor_manager.py delete -id 100001
  • 位置参数“delete”，表示删除一个云主机类型；
  • 参数“-id”支持指定ID查询，返回response，控制台输出response。

api_flavor_manager.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
#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：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
#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}}

基于Python arg用户管理的命令行工具开发

使用已建好的OpenStack Python运维开发环境，在/root目录下创建user_manager.py脚本，完成用户管理功能开发，user_manager.py程序支持命令行带参数执行。

提示说明：Python标准库argparse模块，可以提供命令行参数的解析。

（1）程序支持根据命令行参数，创建1个用户。查询结果，以Json格式输出到控制台。

参考执行实例如下：

1
python3 user_manager.py create --input '{ "name": "user01", "password": "000000", "description": "description" } '
  • 位置参数“create”，表示创建；
  • 参数“-i 或–input”，格式为Json格式文本用户数据。

（2）支持查询给定具体名称的用户查询。

参考执行实例如下：

1
python3 user_manager.py get --name user01 -o user.json
  • 位置参数“get”，表示查询1个用户；
  • 参数“-n 或 –name”支持指定名称查询，类型为string。
  • 参数“-o 或 output”支持查询该用户信息输出到文件，格式为Json格式。

（3）程序支持查询目前admin账号下所有的用户。

参考执行实例如下：

1
python3 user_manager.py getall -o openstack_all_user.yaml
  • 位置参数“getall”，表示查询所有用户；
  • 参数“-o 或–output”支持输出到文件，格式为yaml格式。

（4）支持删除指定的名称的用户。

参考执行实例如下：

1
python3 user_manager.py delete --name user01
  • 位置参数“delete”，表示删除一个用户；返回response，通过控制台输出。
  • 参数“-n或–name”支持指定名称查询，类型为string。

api_user_manager.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
# 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：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
#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}}