init
Some checks failed
Build And Test / build-and-push (push) Failing after 53s

This commit is contained in:
2024-12-13 17:12:03 +09:00
parent a27be94b19
commit ea11832a53
695 changed files with 71766 additions and 1 deletions

View File

@ -0,0 +1,143 @@
import json
import requests
# type : network
# Region :
# 판교: https://kr1-api-instance-infrastructure.nhncloudservice.com
# 평촌: https://kr2-api-instance-infrastructure.nhncloudservice.com
endpoint_cpt_kr1 = "https://kr1-api-instance-infrastructure.nhncloudservice.com"
endpoint_cpt_kr2 = "https://kr2-api-instance-infrastructure.nhncloudservice.com"
endpoint_img_kr1 = "https://kr1-api-image-infrastructure.nhncloudservice.com"
endpoint_img_kr2 = "https://kr2-api-image-infrastructure.nhncloudservice.com"
class ApiCompute:
def __init__(self, region, tenant_id, token):
if region == "kr1":
self.cptUrl = endpoint_cpt_kr1
self.imgUrl = endpoint_img_kr1
elif region == "kr2":
self.cptUrl = endpoint_cpt_kr2
self.imgUrl = endpoint_img_kr2
else:
self.cptUrl = endpoint_cpt_kr2
self.imgUrl = endpoint_img_kr2
self.token = token
self.tenantId = tenant_id
def getFlavorList(self):
url = f"{self.cptUrl}/v2/{self.tenantId}/flavors"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas
def getFlavorId(self, flavor_name):
url = f"{self.cptUrl}/v2/{self.tenantId}/flavors"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)['flavors']
for var in datas:
foo = str(var['name'])
if foo.count(flavor_name) == 1:
return var['id']
def getKeypairList(self):
url = f"{self.cptUrl}/v2/{self.tenantId}/os-keypairs"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas
def getKeypairInfo(self, keypair_name):
url = f"{self.cptUrl}/v2/{self.tenantId}/os-keypairs/{keypair_name}"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas
def InstanceList(self):
pass
def InstanceStatus(self, serverId):
url = f"{self.cptUrl}/v2/{self.tenantId}/servers/{serverId}"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)['server']['status']
return datas
def InstanceInfo(self, serverId):
url = f"{self.cptUrl}/v2/{self.tenantId}/servers/{serverId}"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text) # ['server']['status']
return datas
def InstanceId(self, instance_name):
pass
def deleteInstance(self, instance_id):
pass
def createInstance(self, instance_name, instance_image_id, flavor_id, subnet_id, kepair_name):
url = f"{self.cptUrl}/v2/{self.tenantId}/servers"
parms = {
"server": {
"name": instance_name,
"imageRef": instance_image_id,
"flavorRef": flavor_id,
"networks": [
{
"subnet": f"{subnet_id}",
}
],
# "availability_zone": "kr-pub-b",
"key_name": kepair_name,
"max_count": 1,
"min_count": 1,
"block_device_mapping_v2": [
{
"uuid": f"{instance_image_id}",
"boot_index": 0,
"volume_size": 50,
"device_name": "vda",
"source_type": "image",
"destination_type": "volume",
"delete_on_termination": 1,
}
],
"security_groups": [{"name": "default"}],
}
}
headers = {
"X-Auth-Token": self.token,
}
# print(parms)
response = requests.post(url, json=parms, headers=headers)
datas = json.loads(response.text)['server']
# print("------------------------")
# print(response)
# print("------------------------")
# print(datas)
return datas['id']
def getImgListAll(self):
url = f"{self.imgUrl}/v2/images"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas
def getInstanceImgId(self, img_name):
url = f"{self.imgUrl}/v2/images"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)['images']
for var in datas:
foo = str(var['name'])
if foo.startswith(img_name) and foo.count("Container") == 0:
return var['id']

175
nhncloud/packages/ApiNks.py Normal file
View File

@ -0,0 +1,175 @@
import json
import requests
import yaml
# type : network
# Region :
# 판교: https://kr1-api-kubernetes-infrastructure.nhncloudservice.com
# 평촌: https://kr2-api-kubernetes-infrastructure.nhncloudservice.com
endpoint_kr1 = "https://kr1-api-kubernetes-infrastructure.nhncloudservice.com"
endpoint_kr2 = "https://kr2-api-kubernetes-infrastructure.nhncloudservice.com"
class ApiNks:
def __init__(self, region, token):
if region == "kr1" :
self.region = region
self.url = endpoint_kr1
elif region == "kr2" :
self.region = region
self.url = endpoint_kr2
else:
self.region = region
self.url = endpoint_kr2
self.token = token
def getClusterList(self):
url = f"{self.url}/v1/clusters"
headers={
"X-Auth-Token": self.token,
"Accept": "application/json",
"Content-Type": "application/json",
"OpenStack-API-Version": "container-infra latest"
}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas
def getClusterInfo(self, cluster_name):
url = f"{self.url}/v1/clusters/{cluster_name}"
headers={
"X-Auth-Token": self.token,
"Accept": "application/json",
"Content-Type": "application/json",
"OpenStack-API-Version": "container-infra latest"
}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas
def publicClusterCreate(self, parms_dict):
print(f"리전정보 {self.region}")
if self.region == "kr1" :
node_img = "1213d033-bdf6-4d73-9763-4e8e57c745fb" # ubuntu 20.04
elif self.region == "kr2" :
node_img = "dabb6d10-937d-4952-9ce0-1e576e9164e8" # ubuntu 20.04
else:
node_img = "dabb6d10-937d-4952-9ce0-1e576e9164e8" # ubuntu 20.04
url = f"{self.url}/v1/clusters"
parms = {
"cluster_template_id": "iaas_console",
"create_timeout": 60,
"fixed_network": str(parms_dict["vpc_id"]),
"fixed_subnet": str(parms_dict["subnet_id"]),
"flavor_id": str(parms_dict["instance_type"]),
"keypair": str(parms_dict["key_pair"]),
"labels": {
"availability_zone": str(parms_dict["az-zone"]),
"boot_volume_size": "50",
"boot_volume_type": "General SSD",
"ca_enable": "false",
# "ca_max_node_count": "10",
# "ca_min_node_count": "1",
# "ca_scale_down_delay_after_add": "10", # 샘플은3인데 왜 10일 요구하나
# "ca_scale_down_enable": "true",
# "ca_scale_down_unneeded_time": "3",
# "ca_scale_down_util_thresh": "50",
# "cert_manager_api": "True",
"cert_manager_api": "True",
"clusterautoscale": "nodegroupfeature",
"external_network_id": str(parms_dict["external_network_id"]), # 외부 연결된경우 적용
"external_subnet_id_list": str(parms_dict["external_subnet_id_list"]), # 외부 연결
"kube_tag": str(parms_dict["kubernetes-version"]),
"master_lb_floating_ip_enabled": "True", # external_network, subnet적용시 True가능.
"node_image": node_img,
# "user_script_v2": ""
},
"name": str(parms_dict["cluster_name"]),
"node_count": str(parms_dict["nd_cnt"])
}
headers={
"Accept": "application/json",
"Content-Type": "application/json",
"OpenStack-API-Version": "container-infra latest",
"X-Auth-Token": self.token
}
response = requests.post(url, json=parms, headers=headers)
print(f"debug:{response}")
datas = json.loads(response.text)
# print("------------------------")
# print(response)
# print("------------------------")
print(datas)
return datas
def privateClusterCreate(self, parms_dict):
if self.region == "kr1" :
node_img = "1213d033-bdf6-4d73-9763-4e8e57c745fb" # ubuntu 20.04
elif self.region == "kr2" :
node_img = "dabb6d10-937d-4952-9ce0-1e576e9164e8" # ubuntu 20.04
else:
node_img = "dabb6d10-937d-4952-9ce0-1e576e9164e8" # ubuntu 20.04
url = f"{self.url}/v1/clusters"
parms = {
"cluster_template_id": "iaas_console",
"create_timeout": 60,
"fixed_network": str(parms_dict["vpc_id"]),
"fixed_subnet": str(parms_dict["subnet_id"]),
"flavor_id": str(parms_dict["instance_type"]),
"keypair": str(parms_dict["key_pair"]),
"labels": {
"availability_zone": str(parms_dict["az-zone"]),
"boot_volume_size": "50",
"boot_volume_type": "General SSD",
"ca_enable": "false",
"cert_manager_api": "True",
"clusterautoscale": "nodegroupfeature",
"kube_tag": str(parms_dict["kubernetes-version"]),
"master_lb_floating_ip_enabled": "False", # external_network, subnet적용시 True가능.
"node_image": node_img,
},
"name": str(parms_dict["cluster_name"]),
"node_count": str(parms_dict["nd_cnt"])
}
headers={
"Accept": "application/json",
"Content-Type": "application/json",
"OpenStack-API-Version": "container-infra latest",
"X-Auth-Token": self.token
}
response = requests.post(url, json=parms, headers=headers)
datas = json.loads(response.text)
# print("DEBUG MODE")
# print(response)
# print("DEBUG MODE")
# print(datas)
return datas
def nksDelete(self, cluster_name):
url = f"{self.url}/v1/clusters/{cluster_name}"
headers={
"X-Auth-Token": self.token,
"Accept": "application/json",
"Content-Type": "application/json",
"OpenStack-API-Version": "container-infra latest"
}
response = requests.delete(url, headers=headers)
return
def getClusterConfig(self, cluster_name): # 작성만 하였음 확인 필요.
url = f"{self.url}/v1/clusters/{cluster_name}/config"
headers={
"X-Auth-Token": self.token,
"Accept": "application/json",
"Content-Type": "application/json",
"OpenStack-API-Version": "container-infra latest"
}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas['config']

View File

@ -0,0 +1,95 @@
import requests
import os
endpoint_url_kr1 = "https://kr1-api-object-storage.nhncloudservice.com/v1"
endpoint_url_kr2 = "https://kr2-api-object-storage.nhncloudservice.com/v1"
class ApiStorageObject:
def __init__(self, region, token, storageAccount):
if region == "kr1":
self.storage_url = endpoint_url_kr1
elif region == "kr2":
self.storage_url = endpoint_url_kr2
else:
self.storage_url = endpoint_url_kr2
self.token = token
self.scAccount = storageAccount
def _get_url(self, container=None):
if container == None:
container = ""
else:
container = container
return self.storage_url + "/" + container
def _get_obj_url(self, container=None, object=None):
return '/'.join([self.storage_url, container, object])
def _get_request_header(self):
return {"X-Auth-Token": self.token}
def get_container_list(self):
req_url = str(f"{self._get_url()}{self.scAccount}")
print(f"url: {req_url}")
headers = self._get_request_header()
resp = requests.get(req_url, headers=headers)
# print(resp)
return resp.text.split("\n")
def create(self, container):
req_url = str(f"{self._get_url()}{self.scAccount}/{container}")
print(req_url)
req_header = self._get_request_header()
return requests.put(req_url, headers=req_header)
def remove(self, container):
req_url = str(f"{self._get_url()}{self.scAccount}/{container}")
print(req_url)
req_header = self._get_request_header()
return requests.delete(req_url, headers=req_header)
# public 으로 변경
def set_read_acl(self, container, is_public):
req_url = str(f"{self._get_url()}{self.scAccount}/{container}")
req_header = self._get_request_header()
req_header["X-Container-Read"] = ".r:*" if is_public else ""
return requests.post(req_url, headers=req_header)
def upload(self, container, object, object_path):
req_url = str(f"{self._get_url()}{self.scAccount}/{container}/{object}")
req_header = self._get_request_header()
path = '/'.join([object_path, object])
with open(path, 'rb') as f:
return requests.put(req_url, headers=req_header, data=f.read())
def update(self, container, object):
req_url = str(f"{self._get_url()}{self.scAccount}/{container}/{object}")
print(req_url)
headers={
"X-Auth-Token": self.token,
"Content" : "multipart/form-data",
"X-Delete-At" : "1705985040"
}
return requests.put(req_url, headers=headers)
# 컨테이너 디렉토리 리스트
def _get_list(self, container):
req_url = str(f"{self._get_url()}{self.scAccount}/{container}")
req_header = self._get_request_header()
response = requests.get(req_url, headers=req_header)
return response.text.split('\n')
def get_object_list(self, container):
req_url = str(f"{self._get_url()}{self.scAccount}/{container}")
print(req_url)
return self._get_list(req_url)
def delete(self, containerName, objectPath):
req_url = f"{self._get_url()}{self.scAccount}/{containerName}/{objectPath}"
headers={
"X-Auth-Token": self.token,
}
return requests.delete(req_url, headers=headers)

View File

@ -0,0 +1,42 @@
import requests
import json
# def createToken(url, parms):
def createToken(parms):
url = "https://api-identity-infrastructure.nhncloudservice.com/v2.0/tokens"
response = requests.post(url, json=parms) # 응답요청
datas = json.loads(response.text)
# print(datas)
if "error" in datas:
errDict = {
'state' : 'err',
'code' : datas['error']['code'],
'errTitle' : datas['error']['title'],
'errMsg' : datas['error']['message']
}
return errDict
else:
getDict = {
'state' : 'steady',
'tokenValue' : datas['access']['token']['id']
}
# WhatToken = getDict['tokenValue']
return getDict
if __name__ == "__main__":
# 예제 내용
url = "https://api-identity-infrastructure.nhncloudservice.com/v2.0/tokens"
parms = {
"auth": {
"tenantId": "04a2c5b7de7e4d66b970ad950081a7c3", # kr2
"passwordCredentials": {
# "username": "sdjo@injeinc.co.kr",
# "password": "ijinc123!"
}
}
}
print(createToken(url, parms))

229
nhncloud/packages/ApiVpc.py Normal file
View File

@ -0,0 +1,229 @@
import json
import requests
# type : network
# Region :
# 판교: https://kr1-api-network-infrastructure.nhncloudservice.com
# 평촌: https://kr2-api-network-infrastructure.nhncloudservice.com
endpoint_vpc_kr1 = "https://kr1-api-network-infrastructure.nhncloudservice.com"
endpoint_vpc_kr2 = "https://kr2-api-network-infrastructure.nhncloudservice.com"
endpoint_floating_kr1 = "https://kr1-api-network-infrastructure.nhncloudservice.com"
endpoint_floating_kr2 = "https://kr2-api-network-infrastructure.nhncloudservice.com"
class ApiVpc:
def __init__(self, region, token):
if region == "kr1":
self.vpcUrl = endpoint_vpc_kr1
self.floatingUrl = endpoint_floating_kr1
elif region == "kr2":
self.vpcUrl = endpoint_vpc_kr2
self.floatingUrl = endpoint_floating_kr2
else:
self.vpcUrl = endpoint_vpc_kr2
self.floatingUrl = endpoint_floating_kr2
self.token = token
def vpcList(self):
url = f"{self.vpcUrl}/v2.0/vpcs"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas
def vpcInfo(self, vpc_id):
url = f"{self.vpcUrl}/v2.0/vpcs/{vpc_id}"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas
def getVpcId(self, vpc_name):
vpcName = vpc_name
url = f"{self.vpcUrl}/v2.0/vpcs"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)["vpcs"]
print(f"ss:{datas}")
i = len(datas) - 1
for var in reversed(datas):
foo = str(var["name"])
if foo.startswith(vpcName):
break
else:
if i == 0:
break
else:
i -= 1
return datas[i]["id"]
def vpcCreate(self, name, cidrv4):
url = f"{self.vpcUrl}/v2.0/vpcs"
parms = {"vpc": {"name": name, "cidrv4": cidrv4}}
headers = {"X-Auth-Token": self.token}
response = requests.post(url, json=parms, headers=headers)
datas = json.loads(response.text)
return datas["vpc"]["id"]
def vpcDelete(self, vpc_id):
url = f"{self.vpcUrl}/v2.0/vpcs/{vpc_id}"
headers = {"X-Auth-Token": self.token}
response = requests.delete(url, headers=headers)
return
def subnetList(self):
url = f"{self.vpcUrl}/v2.0/vpcsubnets"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas
def getSubnetId(self, subnetName):
url = f"{self.vpcUrl}/v2.0/vpcsubnets"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)["vpcsubnets"]
for var in datas:
foo = str(var["name"])
if foo.startswith(subnetName) and foo.count("rt") == 0:
return var["id"]
def subnetCreate(self, vpc_id, cidr, name):
url = f"{self.vpcUrl}/v2.0/vpcsubnets"
parms = {"vpcsubnet": {"vpc_id": vpc_id, "cidr": cidr, "name": name}}
headers = {"X-Auth-Token": self.token}
response = requests.post(url, json=parms, headers=headers)
datas = json.loads(response.text)
return datas["vpcsubnet"]["id"]
def subnetDelete(self, subnet_id):
url = f"{self.vpcUrl}/v2.0/vpcsubnets/{subnet_id}"
headers = {"X-Auth-Token": self.token}
response = requests.delete(url, headers=headers)
return
def routingList(self):
# url = f"{self.vpcUrl}/v2.0/routingtables"
url = f"{self.vpcUrl}/v2.0/routingtables?detail=true"
# parms = {
# "default_table": "true"
# }
headers = {
"X-Auth-Token": self.token,
#
}
response = requests.get(url, headers=headers)
# response = requests.post(url, query={"detail":"true"}, headers=headers)
datas = json.loads(response.text)
return datas
def getRoutingDefault(self):
url = f"{self.vpcUrl}/v2.0/routingtables"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas["routingtables"][0]["id"]
def routingCreate(self, name, vpc_id, distributed="true"):
url = f"{self.vpcUrl}/v2.0/routingtables"
parms = {
"routingtable": {"name": name, "vpc_id": vpc_id, "distributed": distributed}
}
headers = {"X-Auth-Token": self.token}
response = requests.post(url, json=parms, headers=headers)
datas = json.loads(response.text)
return datas["routingtable"]["id"]
def routingDelete(self, routingtable_Id):
url = f"{self.vpcUrl}/v2.0/routingtables/{routingtable_Id}"
headers = {"X-Auth-Token": self.token}
response = requests.delete(url, headers=headers)
return
def setDefaultRouting(self, routingtable_Id):
url = f"{self.vpcUrl}/v2.0/routingtables/{routingtable_Id}/set_as_default"
headers = {"X-Auth-Token": self.token}
response = requests.put(url, headers=headers)
datas = json.loads(response.text)
return datas
def setIgwToRoutingTable(self, routingtable_Id, igw_id):
url = f"{self.vpcUrl}/v2.0/routingtables/{routingtable_Id}/attach_gateway"
parms = {"gateway_id": igw_id}
headers = {"X-Auth-Token": self.token}
response = requests.put(url, json=parms, headers=headers)
datas = json.loads(response.text)
return datas
def detachIgwFromRoutingTable(self, routingtable_Id):
url = f"{self.vpcUrl}/v2.0/routingtables/{routingtable_Id}/detach_gateway"
headers = {"X-Auth-Token": self.token}
response = requests.put(url, headers=headers)
datas = json.loads(response.text)
return datas
def attachRoutingToSubnet(self, subnet_Id, routing_id):
url = f"{self.vpcUrl}/v2.0/vpcsubnets/{subnet_Id}/attach_routingtable"
parms = {"routingtable_id": routing_id}
headers = {"X-Auth-Token": self.token}
response = requests.put(url, json=parms, headers=headers)
datas = json.loads(response.text)
return datas
def getExternalNetworkId(self):
url = f"{self.vpcUrl}/v2.0/networks?router:external=True"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas
def createFloatingIp(self, external_id):
url = f"{self.floatingUrl}/v2.0/floatingips"
headers = {"X-Auth-Token": self.token}
parms = {
"floatingip": {
"floating_network_id": external_id,
# "port_id": ""
}
}
response = requests.post(url, json=parms, headers=headers)
# datas = json.loads(response.text)["floatingip"]["floating_ip_address"]
datas = json.loads(response.text)["floatingip"]["id"]
return datas
def attachFloatingIpToDeviceNic(self, floating_ip_id, device_nic_id):
url = f"{self.floatingUrl}/v2.0/floatingips/{floating_ip_id}"
parms = {"floatingip": {"port_id": device_nic_id}}
headers = {"X-Auth-Token": self.token}
response = requests.put(url, json=parms, headers=headers)
print(response)
datas = json.loads(response.text)
return datas
def nicList(self):
url = f"{self.vpcUrl}/v2.0/ports"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)
return datas
def nicId(self, network_id):
url = f"{self.vpcUrl}/v2.0/ports"
headers = {"X-Auth-Token": self.token}
response = requests.get(url, headers=headers)
datas = json.loads(response.text)["ports"]
for var in datas:
foo = str(var["device_id"])
if foo.startswith(network_id):
return var["id"]