v0.0.8 | CORS 설정에 X-NHN-Appkey 헤더 허용 추가
All checks were successful
Build And Test / build-and-push (push) Successful in 36s
All checks were successful
Build And Test / build-and-push (push) Successful in 36s
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@ -3,6 +3,7 @@ from .compute import ApiCompute
|
||||
from .vpc import ApiVpc
|
||||
from .nks import ApiNks
|
||||
from .storage import ApiStorageObject
|
||||
from .dnsplus import ApiDnsPlus
|
||||
|
||||
__all__ = [
|
||||
"NHNCloudToken",
|
||||
@ -10,4 +11,5 @@ __all__ = [
|
||||
"ApiVpc",
|
||||
"ApiNks",
|
||||
"ApiStorageObject",
|
||||
"ApiDnsPlus",
|
||||
]
|
||||
|
||||
@ -46,6 +46,9 @@ class NHNCloudEndpoints:
|
||||
STORAGE_KR1 = "https://kr1-api-object-storage.nhncloudservice.com/v1"
|
||||
STORAGE_KR2 = "https://kr2-api-object-storage.nhncloudservice.com/v1"
|
||||
|
||||
# DNS Plus (글로벌 서비스 - 리전 무관)
|
||||
DNSPLUS = "https://dnsplus.api.nhncloudservice.com"
|
||||
|
||||
|
||||
class NHNCloudAPIError(Exception):
|
||||
"""NHN Cloud API 에러"""
|
||||
@ -163,3 +166,7 @@ class BaseAPI:
|
||||
def _delete(self, url: str, **kwargs) -> dict:
|
||||
"""DELETE 요청"""
|
||||
return self._request("DELETE", url, **kwargs)
|
||||
|
||||
def _patch(self, url: str, json_data: Optional[dict] = None, **kwargs) -> dict:
|
||||
"""PATCH 요청"""
|
||||
return self._request("PATCH", url, json_data=json_data, **kwargs)
|
||||
|
||||
911
nhn/packages/dnsplus.py
Normal file
911
nhn/packages/dnsplus.py
Normal file
@ -0,0 +1,911 @@
|
||||
"""
|
||||
NHN Cloud DNS Plus API Module
|
||||
|
||||
DNS Zone, 레코드 세트, GSLB, Pool, Health Check 관리
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Optional, List
|
||||
|
||||
import requests
|
||||
|
||||
from .base import NHNCloudEndpoints, NHNCloudAPIError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ApiDnsPlus:
|
||||
"""NHN Cloud DNS Plus API 클래스"""
|
||||
|
||||
DEFAULT_TIMEOUT = 30
|
||||
|
||||
def __init__(self, appkey: str):
|
||||
"""
|
||||
Args:
|
||||
appkey: NHN Cloud 앱키
|
||||
"""
|
||||
self.appkey = appkey
|
||||
self.base_url = f"{NHNCloudEndpoints.DNSPLUS}/dnsplus/v1.0/appkeys/{appkey}"
|
||||
self._session = requests.Session()
|
||||
|
||||
def _get_headers(self) -> dict:
|
||||
"""DNS Plus API 헤더"""
|
||||
return {
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
}
|
||||
|
||||
def _request(
|
||||
self,
|
||||
method: str,
|
||||
url: str,
|
||||
params: Optional[dict] = None,
|
||||
json_data: Optional[dict] = None,
|
||||
timeout: Optional[int] = None,
|
||||
) -> dict:
|
||||
"""HTTP 요청 실행"""
|
||||
logger.info(f"[DnsPlus] 요청 시작 - method={method}, url={url}")
|
||||
if params:
|
||||
logger.info(f"[DnsPlus] 요청 파라미터 - params={params}")
|
||||
if json_data:
|
||||
logger.info(f"[DnsPlus] 요청 바디 - json={json_data}")
|
||||
|
||||
try:
|
||||
response = self._session.request(
|
||||
method=method,
|
||||
url=url,
|
||||
params=params,
|
||||
json=json_data,
|
||||
headers=self._get_headers(),
|
||||
timeout=timeout or self.DEFAULT_TIMEOUT,
|
||||
)
|
||||
|
||||
logger.info(f"[DnsPlus] 응답 수신 - status_code={response.status_code}")
|
||||
|
||||
if response.text:
|
||||
result = response.json()
|
||||
# DNS Plus API는 header.isSuccessful로 성공 여부 판단
|
||||
if "header" in result and not result["header"].get("isSuccessful", True):
|
||||
error_msg = result["header"].get("resultMessage", "알 수 없는 오류")
|
||||
error_code = result["header"].get("resultCode", response.status_code)
|
||||
logger.error(f"[DnsPlus] API 오류 - code={error_code}, message={error_msg}")
|
||||
raise NHNCloudAPIError(message=error_msg, code=error_code, details=result)
|
||||
return result
|
||||
return {}
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
logger.error(f"[DnsPlus] 타임아웃 - url={url}")
|
||||
raise NHNCloudAPIError("요청 시간이 초과되었습니다.", code=408)
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
logger.error(f"[DnsPlus] 연결 오류 - url={url}, error={e}")
|
||||
raise NHNCloudAPIError("서버에 연결할 수 없습니다.", code=503)
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.error(f"[DnsPlus] 요청 오류 - url={url}, error={e}")
|
||||
raise NHNCloudAPIError(f"요청 중 오류가 발생했습니다: {e}")
|
||||
|
||||
def _get(self, url: str, params: Optional[dict] = None) -> dict:
|
||||
return self._request("GET", url, params=params)
|
||||
|
||||
def _post(self, url: str, json_data: Optional[dict] = None) -> dict:
|
||||
return self._request("POST", url, json_data=json_data)
|
||||
|
||||
def _put(self, url: str, json_data: Optional[dict] = None) -> dict:
|
||||
return self._request("PUT", url, json_data=json_data)
|
||||
|
||||
def _delete(self, url: str, params: Optional[dict] = None) -> dict:
|
||||
return self._request("DELETE", url, params=params)
|
||||
|
||||
# ==================== DNS Zone ====================
|
||||
|
||||
def get_zone_list(
|
||||
self,
|
||||
zone_id_list: Optional[List[str]] = None,
|
||||
zone_status_list: Optional[List[str]] = None,
|
||||
search_zone_name: Optional[str] = None,
|
||||
page: int = 1,
|
||||
limit: int = 50,
|
||||
) -> dict:
|
||||
"""
|
||||
DNS Zone 목록 조회
|
||||
|
||||
Args:
|
||||
zone_id_list: Zone ID 목록
|
||||
zone_status_list: Zone 상태 목록 (USE, DISUSE)
|
||||
search_zone_name: 검색할 Zone 이름
|
||||
page: 페이지 번호
|
||||
limit: 페이지당 항목 수
|
||||
|
||||
Returns:
|
||||
dict: Zone 목록
|
||||
"""
|
||||
url = f"{self.base_url}/zones"
|
||||
params = {"page": page, "limit": limit}
|
||||
|
||||
if zone_id_list:
|
||||
params["zoneIdList"] = ",".join(zone_id_list)
|
||||
if zone_status_list:
|
||||
params["zoneStatusList"] = ",".join(zone_status_list)
|
||||
if search_zone_name:
|
||||
params["searchZoneName"] = search_zone_name
|
||||
|
||||
return self._get(url, params)
|
||||
|
||||
def get_zone(self, zone_id: str) -> dict:
|
||||
"""
|
||||
DNS Zone 상세 조회
|
||||
|
||||
Args:
|
||||
zone_id: Zone ID
|
||||
|
||||
Returns:
|
||||
dict: Zone 상세 정보
|
||||
"""
|
||||
url = f"{self.base_url}/zones"
|
||||
params = {"zoneIdList": zone_id}
|
||||
result = self._get(url, params)
|
||||
zones = result.get("zoneList", [])
|
||||
if zones:
|
||||
return {"zone": zones[0]}
|
||||
raise NHNCloudAPIError(f"Zone을 찾을 수 없습니다: {zone_id}", code=404)
|
||||
|
||||
def create_zone(self, zone_name: str, description: Optional[str] = None) -> dict:
|
||||
"""
|
||||
DNS Zone 생성
|
||||
|
||||
Args:
|
||||
zone_name: Zone 이름 (도메인, 예: example.com.)
|
||||
description: 설명
|
||||
|
||||
Returns:
|
||||
dict: 생성된 Zone 정보
|
||||
"""
|
||||
url = f"{self.base_url}/zones"
|
||||
payload = {
|
||||
"zone": {
|
||||
"zoneName": zone_name,
|
||||
}
|
||||
}
|
||||
if description:
|
||||
payload["zone"]["description"] = description
|
||||
|
||||
logger.info(f"DNS Zone 생성 요청: {zone_name}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def update_zone(self, zone_id: str, description: str) -> dict:
|
||||
"""
|
||||
DNS Zone 수정
|
||||
|
||||
Args:
|
||||
zone_id: Zone ID
|
||||
description: 설명
|
||||
|
||||
Returns:
|
||||
dict: 수정된 Zone 정보
|
||||
"""
|
||||
url = f"{self.base_url}/zones/{zone_id}"
|
||||
payload = {
|
||||
"zone": {
|
||||
"description": description,
|
||||
}
|
||||
}
|
||||
|
||||
logger.info(f"DNS Zone 수정 요청: {zone_id}")
|
||||
return self._put(url, payload)
|
||||
|
||||
def delete_zones(self, zone_id_list: List[str]) -> dict:
|
||||
"""
|
||||
DNS Zone 삭제 (비동기)
|
||||
|
||||
Args:
|
||||
zone_id_list: 삭제할 Zone ID 목록
|
||||
|
||||
Returns:
|
||||
dict: 삭제 결과
|
||||
"""
|
||||
url = f"{self.base_url}/zones/async"
|
||||
params = {"zoneIdList": ",".join(zone_id_list)}
|
||||
|
||||
logger.info(f"DNS Zone 삭제 요청: {zone_id_list}")
|
||||
return self._delete(url, params)
|
||||
|
||||
# ==================== Record Set ====================
|
||||
|
||||
def get_recordset_list(
|
||||
self,
|
||||
zone_id: str,
|
||||
recordset_id_list: Optional[List[str]] = None,
|
||||
recordset_type_list: Optional[List[str]] = None,
|
||||
search_recordset_name: Optional[str] = None,
|
||||
page: int = 1,
|
||||
limit: int = 50,
|
||||
) -> dict:
|
||||
"""
|
||||
레코드 세트 목록 조회
|
||||
|
||||
Args:
|
||||
zone_id: Zone ID
|
||||
recordset_id_list: 레코드 세트 ID 목록
|
||||
recordset_type_list: 레코드 타입 목록 (A, AAAA, CNAME, MX, TXT, NS, SRV, etc.)
|
||||
search_recordset_name: 검색할 레코드 세트 이름
|
||||
page: 페이지 번호
|
||||
limit: 페이지당 항목 수
|
||||
|
||||
Returns:
|
||||
dict: 레코드 세트 목록
|
||||
"""
|
||||
url = f"{self.base_url}/zones/{zone_id}/recordsets"
|
||||
params = {"page": page, "limit": limit}
|
||||
|
||||
if recordset_id_list:
|
||||
params["recordsetIdList"] = ",".join(recordset_id_list)
|
||||
if recordset_type_list:
|
||||
params["recordsetTypeList"] = ",".join(recordset_type_list)
|
||||
if search_recordset_name:
|
||||
params["searchRecordsetName"] = search_recordset_name
|
||||
|
||||
return self._get(url, params)
|
||||
|
||||
def get_recordset(self, zone_id: str, recordset_id: str) -> dict:
|
||||
"""
|
||||
레코드 세트 상세 조회
|
||||
|
||||
Args:
|
||||
zone_id: Zone ID
|
||||
recordset_id: 레코드 세트 ID
|
||||
|
||||
Returns:
|
||||
dict: 레코드 세트 상세 정보
|
||||
"""
|
||||
url = f"{self.base_url}/zones/{zone_id}/recordsets"
|
||||
params = {"recordsetIdList": recordset_id}
|
||||
result = self._get(url, params)
|
||||
recordsets = result.get("recordsetList", [])
|
||||
if recordsets:
|
||||
return {"recordset": recordsets[0]}
|
||||
raise NHNCloudAPIError(f"레코드 세트를 찾을 수 없습니다: {recordset_id}", code=404)
|
||||
|
||||
def create_recordset(
|
||||
self,
|
||||
zone_id: str,
|
||||
recordset_name: str,
|
||||
recordset_type: str,
|
||||
recordset_ttl: int,
|
||||
record_list: List[dict],
|
||||
) -> dict:
|
||||
"""
|
||||
레코드 세트 생성
|
||||
|
||||
Args:
|
||||
zone_id: Zone ID
|
||||
recordset_name: 레코드 세트 이름
|
||||
recordset_type: 레코드 타입 (A, AAAA, CNAME, MX, TXT, NS, SRV, etc.)
|
||||
recordset_ttl: TTL (초)
|
||||
record_list: 레코드 목록 [{"recordContent": "...", "recordDisabled": False}]
|
||||
|
||||
Returns:
|
||||
dict: 생성된 레코드 세트 정보
|
||||
"""
|
||||
url = f"{self.base_url}/zones/{zone_id}/recordsets"
|
||||
payload = {
|
||||
"recordset": {
|
||||
"recordsetName": recordset_name,
|
||||
"recordsetType": recordset_type,
|
||||
"recordsetTtl": recordset_ttl,
|
||||
"recordList": record_list,
|
||||
}
|
||||
}
|
||||
|
||||
logger.info(f"레코드 세트 생성 요청: {recordset_name} ({recordset_type})")
|
||||
return self._post(url, payload)
|
||||
|
||||
def create_recordset_bulk(
|
||||
self,
|
||||
zone_id: str,
|
||||
recordset_list: List[dict],
|
||||
) -> dict:
|
||||
"""
|
||||
레코드 세트 대량 생성
|
||||
|
||||
Args:
|
||||
zone_id: Zone ID
|
||||
recordset_list: 레코드 세트 목록 (최대 2,000개)
|
||||
|
||||
Returns:
|
||||
dict: 생성된 레코드 세트 정보
|
||||
"""
|
||||
url = f"{self.base_url}/zones/{zone_id}/recordsets/list"
|
||||
payload = {"recordsetList": recordset_list}
|
||||
|
||||
logger.info(f"레코드 세트 대량 생성 요청: {len(recordset_list)}개")
|
||||
return self._post(url, payload)
|
||||
|
||||
def update_recordset(
|
||||
self,
|
||||
zone_id: str,
|
||||
recordset_id: str,
|
||||
recordset_type: str,
|
||||
recordset_ttl: int,
|
||||
record_list: List[dict],
|
||||
) -> dict:
|
||||
"""
|
||||
레코드 세트 수정
|
||||
|
||||
Args:
|
||||
zone_id: Zone ID
|
||||
recordset_id: 레코드 세트 ID
|
||||
recordset_type: 레코드 타입
|
||||
recordset_ttl: TTL (초)
|
||||
record_list: 레코드 목록
|
||||
|
||||
Returns:
|
||||
dict: 수정된 레코드 세트 정보
|
||||
"""
|
||||
url = f"{self.base_url}/zones/{zone_id}/recordsets/{recordset_id}"
|
||||
payload = {
|
||||
"recordset": {
|
||||
"recordsetType": recordset_type,
|
||||
"recordsetTtl": recordset_ttl,
|
||||
"recordList": record_list,
|
||||
}
|
||||
}
|
||||
|
||||
logger.info(f"레코드 세트 수정 요청: {recordset_id}")
|
||||
return self._put(url, payload)
|
||||
|
||||
def delete_recordsets(self, zone_id: str, recordset_id_list: List[str]) -> dict:
|
||||
"""
|
||||
레코드 세트 삭제
|
||||
|
||||
Args:
|
||||
zone_id: Zone ID
|
||||
recordset_id_list: 삭제할 레코드 세트 ID 목록
|
||||
|
||||
Returns:
|
||||
dict: 삭제 결과
|
||||
"""
|
||||
url = f"{self.base_url}/zones/{zone_id}/recordsets"
|
||||
params = {"recordsetIdList": ",".join(recordset_id_list)}
|
||||
|
||||
logger.info(f"레코드 세트 삭제 요청: {recordset_id_list}")
|
||||
return self._delete(url, params)
|
||||
|
||||
# ==================== GSLB ====================
|
||||
|
||||
def get_gslb_list(
|
||||
self,
|
||||
gslb_id_list: Optional[List[str]] = None,
|
||||
search_gslb_name: Optional[str] = None,
|
||||
gslb_domain: Optional[str] = None,
|
||||
page: int = 1,
|
||||
limit: int = 50,
|
||||
) -> dict:
|
||||
"""
|
||||
GSLB 목록 조회
|
||||
|
||||
Args:
|
||||
gslb_id_list: GSLB ID 목록
|
||||
search_gslb_name: 검색할 GSLB 이름
|
||||
gslb_domain: GSLB 도메인
|
||||
page: 페이지 번호
|
||||
limit: 페이지당 항목 수
|
||||
|
||||
Returns:
|
||||
dict: GSLB 목록
|
||||
"""
|
||||
url = f"{self.base_url}/gslbs"
|
||||
params = {"page": page, "limit": limit}
|
||||
|
||||
if gslb_id_list:
|
||||
params["gslbIdList"] = ",".join(gslb_id_list)
|
||||
if search_gslb_name:
|
||||
params["searchGslbName"] = search_gslb_name
|
||||
if gslb_domain:
|
||||
params["gslbDomain"] = gslb_domain
|
||||
|
||||
return self._get(url, params)
|
||||
|
||||
def get_gslb(self, gslb_id: str) -> dict:
|
||||
"""
|
||||
GSLB 상세 조회
|
||||
|
||||
Args:
|
||||
gslb_id: GSLB ID
|
||||
|
||||
Returns:
|
||||
dict: GSLB 상세 정보
|
||||
"""
|
||||
url = f"{self.base_url}/gslbs"
|
||||
params = {"gslbIdList": gslb_id}
|
||||
result = self._get(url, params)
|
||||
gslbs = result.get("gslbList", [])
|
||||
if gslbs:
|
||||
return {"gslb": gslbs[0]}
|
||||
raise NHNCloudAPIError(f"GSLB를 찾을 수 없습니다: {gslb_id}", code=404)
|
||||
|
||||
def create_gslb(
|
||||
self,
|
||||
gslb_name: str,
|
||||
gslb_ttl: int,
|
||||
gslb_routing_rule: str,
|
||||
gslb_disabled: bool = False,
|
||||
connected_pool_list: Optional[List[dict]] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
GSLB 생성
|
||||
|
||||
Args:
|
||||
gslb_name: GSLB 이름
|
||||
gslb_ttl: TTL (초)
|
||||
gslb_routing_rule: 라우팅 규칙 (FAILOVER, RANDOM, GEOLOCATION)
|
||||
gslb_disabled: 비활성화 여부
|
||||
connected_pool_list: 연결된 Pool 목록
|
||||
|
||||
Returns:
|
||||
dict: 생성된 GSLB 정보
|
||||
"""
|
||||
url = f"{self.base_url}/gslbs"
|
||||
payload = {
|
||||
"gslb": {
|
||||
"gslbName": gslb_name,
|
||||
"gslbTtl": gslb_ttl,
|
||||
"gslbRoutingRule": gslb_routing_rule,
|
||||
"gslbDisabled": gslb_disabled,
|
||||
}
|
||||
}
|
||||
if connected_pool_list:
|
||||
payload["gslb"]["connectedPoolList"] = connected_pool_list
|
||||
|
||||
logger.info(f"GSLB 생성 요청: {gslb_name}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def update_gslb(
|
||||
self,
|
||||
gslb_id: str,
|
||||
gslb_name: Optional[str] = None,
|
||||
gslb_ttl: Optional[int] = None,
|
||||
gslb_routing_rule: Optional[str] = None,
|
||||
gslb_disabled: Optional[bool] = None,
|
||||
connected_pool_list: Optional[List[dict]] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
GSLB 수정
|
||||
|
||||
Args:
|
||||
gslb_id: GSLB ID
|
||||
gslb_name: GSLB 이름
|
||||
gslb_ttl: TTL (초)
|
||||
gslb_routing_rule: 라우팅 규칙
|
||||
gslb_disabled: 비활성화 여부
|
||||
connected_pool_list: 연결된 Pool 목록
|
||||
|
||||
Returns:
|
||||
dict: 수정된 GSLB 정보
|
||||
"""
|
||||
url = f"{self.base_url}/gslbs/{gslb_id}"
|
||||
gslb_data = {}
|
||||
|
||||
if gslb_name is not None:
|
||||
gslb_data["gslbName"] = gslb_name
|
||||
if gslb_ttl is not None:
|
||||
gslb_data["gslbTtl"] = gslb_ttl
|
||||
if gslb_routing_rule is not None:
|
||||
gslb_data["gslbRoutingRule"] = gslb_routing_rule
|
||||
if gslb_disabled is not None:
|
||||
gslb_data["gslbDisabled"] = gslb_disabled
|
||||
if connected_pool_list is not None:
|
||||
gslb_data["connectedPoolList"] = connected_pool_list
|
||||
|
||||
payload = {"gslb": gslb_data}
|
||||
|
||||
logger.info(f"GSLB 수정 요청: {gslb_id}")
|
||||
return self._put(url, payload)
|
||||
|
||||
def delete_gslbs(self, gslb_id_list: List[str]) -> dict:
|
||||
"""
|
||||
GSLB 삭제
|
||||
|
||||
Args:
|
||||
gslb_id_list: 삭제할 GSLB ID 목록
|
||||
|
||||
Returns:
|
||||
dict: 삭제 결과
|
||||
"""
|
||||
url = f"{self.base_url}/gslbs"
|
||||
params = {"gslbIdList": ",".join(gslb_id_list)}
|
||||
|
||||
logger.info(f"GSLB 삭제 요청: {gslb_id_list}")
|
||||
return self._delete(url, params)
|
||||
|
||||
def connect_pool(
|
||||
self,
|
||||
gslb_id: str,
|
||||
pool_id: str,
|
||||
connected_pool_order: int,
|
||||
connected_pool_region_content: Optional[str] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
GSLB에 Pool 연결
|
||||
|
||||
Args:
|
||||
gslb_id: GSLB ID
|
||||
pool_id: Pool ID
|
||||
connected_pool_order: 연결 순서 (우선순위)
|
||||
connected_pool_region_content: 지역 콘텐츠 (GEOLOCATION 규칙 사용 시)
|
||||
|
||||
Returns:
|
||||
dict: 연결 결과
|
||||
"""
|
||||
url = f"{self.base_url}/gslbs/{gslb_id}/connected-pools/{pool_id}"
|
||||
payload = {
|
||||
"connectedPool": {
|
||||
"connectedPoolOrder": connected_pool_order,
|
||||
}
|
||||
}
|
||||
if connected_pool_region_content:
|
||||
payload["connectedPool"]["connectedPoolRegionContent"] = connected_pool_region_content
|
||||
|
||||
logger.info(f"Pool 연결 요청: GSLB={gslb_id}, Pool={pool_id}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def update_connected_pool(
|
||||
self,
|
||||
gslb_id: str,
|
||||
pool_id: str,
|
||||
connected_pool_order: int,
|
||||
connected_pool_region_content: Optional[str] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
연결된 Pool 수정
|
||||
|
||||
Args:
|
||||
gslb_id: GSLB ID
|
||||
pool_id: Pool ID
|
||||
connected_pool_order: 연결 순서 (우선순위)
|
||||
connected_pool_region_content: 지역 콘텐츠
|
||||
|
||||
Returns:
|
||||
dict: 수정 결과
|
||||
"""
|
||||
url = f"{self.base_url}/gslbs/{gslb_id}/connected-pools/{pool_id}"
|
||||
payload = {
|
||||
"connectedPool": {
|
||||
"connectedPoolOrder": connected_pool_order,
|
||||
}
|
||||
}
|
||||
if connected_pool_region_content:
|
||||
payload["connectedPool"]["connectedPoolRegionContent"] = connected_pool_region_content
|
||||
|
||||
logger.info(f"연결된 Pool 수정 요청: GSLB={gslb_id}, Pool={pool_id}")
|
||||
return self._put(url, payload)
|
||||
|
||||
def disconnect_pools(self, gslb_id: str, pool_id_list: List[str]) -> dict:
|
||||
"""
|
||||
GSLB에서 Pool 연결 해제
|
||||
|
||||
Args:
|
||||
gslb_id: GSLB ID
|
||||
pool_id_list: 연결 해제할 Pool ID 목록
|
||||
|
||||
Returns:
|
||||
dict: 연결 해제 결과
|
||||
"""
|
||||
url = f"{self.base_url}/gslbs/{gslb_id}/connected-pools"
|
||||
params = {"poolIdList": ",".join(pool_id_list)}
|
||||
|
||||
logger.info(f"Pool 연결 해제 요청: GSLB={gslb_id}, Pools={pool_id_list}")
|
||||
return self._delete(url, params)
|
||||
|
||||
# ==================== Pool ====================
|
||||
|
||||
def get_pool_list(
|
||||
self,
|
||||
pool_id_list: Optional[List[str]] = None,
|
||||
search_pool_name: Optional[str] = None,
|
||||
health_check_id: Optional[str] = None,
|
||||
page: int = 1,
|
||||
limit: int = 50,
|
||||
) -> dict:
|
||||
"""
|
||||
Pool 목록 조회
|
||||
|
||||
Args:
|
||||
pool_id_list: Pool ID 목록
|
||||
search_pool_name: 검색할 Pool 이름
|
||||
health_check_id: Health Check ID
|
||||
page: 페이지 번호
|
||||
limit: 페이지당 항목 수
|
||||
|
||||
Returns:
|
||||
dict: Pool 목록
|
||||
"""
|
||||
url = f"{self.base_url}/pools"
|
||||
params = {"page": page, "limit": limit}
|
||||
|
||||
if pool_id_list:
|
||||
params["poolIdList"] = ",".join(pool_id_list)
|
||||
if search_pool_name:
|
||||
params["searchPoolName"] = search_pool_name
|
||||
if health_check_id:
|
||||
params["healthCheckId"] = health_check_id
|
||||
|
||||
return self._get(url, params)
|
||||
|
||||
def get_pool(self, pool_id: str) -> dict:
|
||||
"""
|
||||
Pool 상세 조회
|
||||
|
||||
Args:
|
||||
pool_id: Pool ID
|
||||
|
||||
Returns:
|
||||
dict: Pool 상세 정보
|
||||
"""
|
||||
url = f"{self.base_url}/pools"
|
||||
params = {"poolIdList": pool_id}
|
||||
result = self._get(url, params)
|
||||
pools = result.get("poolList", [])
|
||||
if pools:
|
||||
return {"pool": pools[0]}
|
||||
raise NHNCloudAPIError(f"Pool을 찾을 수 없습니다: {pool_id}", code=404)
|
||||
|
||||
def create_pool(
|
||||
self,
|
||||
pool_name: str,
|
||||
endpoint_list: List[dict],
|
||||
pool_disabled: bool = False,
|
||||
health_check_id: Optional[str] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Pool 생성
|
||||
|
||||
Args:
|
||||
pool_name: Pool 이름
|
||||
endpoint_list: 엔드포인트 목록
|
||||
[{"endpointAddress": "1.1.1.1", "endpointWeight": 1.0, "endpointDisabled": False}]
|
||||
pool_disabled: 비활성화 여부
|
||||
health_check_id: Health Check ID
|
||||
|
||||
Returns:
|
||||
dict: 생성된 Pool 정보
|
||||
"""
|
||||
url = f"{self.base_url}/pools"
|
||||
payload = {
|
||||
"pool": {
|
||||
"poolName": pool_name,
|
||||
"poolDisabled": pool_disabled,
|
||||
"endpointList": endpoint_list,
|
||||
}
|
||||
}
|
||||
if health_check_id:
|
||||
payload["pool"]["healthCheckId"] = health_check_id
|
||||
|
||||
logger.info(f"Pool 생성 요청: {pool_name}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def update_pool(
|
||||
self,
|
||||
pool_id: str,
|
||||
pool_name: Optional[str] = None,
|
||||
endpoint_list: Optional[List[dict]] = None,
|
||||
pool_disabled: Optional[bool] = None,
|
||||
health_check_id: Optional[str] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Pool 수정
|
||||
|
||||
Args:
|
||||
pool_id: Pool ID
|
||||
pool_name: Pool 이름
|
||||
endpoint_list: 엔드포인트 목록
|
||||
pool_disabled: 비활성화 여부
|
||||
health_check_id: Health Check ID
|
||||
|
||||
Returns:
|
||||
dict: 수정된 Pool 정보
|
||||
"""
|
||||
url = f"{self.base_url}/pools/{pool_id}"
|
||||
pool_data = {}
|
||||
|
||||
if pool_name is not None:
|
||||
pool_data["poolName"] = pool_name
|
||||
if endpoint_list is not None:
|
||||
pool_data["endpointList"] = endpoint_list
|
||||
if pool_disabled is not None:
|
||||
pool_data["poolDisabled"] = pool_disabled
|
||||
if health_check_id is not None:
|
||||
pool_data["healthCheckId"] = health_check_id
|
||||
|
||||
payload = {"pool": pool_data}
|
||||
|
||||
logger.info(f"Pool 수정 요청: {pool_id}")
|
||||
return self._put(url, payload)
|
||||
|
||||
def delete_pools(self, pool_id_list: List[str]) -> dict:
|
||||
"""
|
||||
Pool 삭제
|
||||
|
||||
Args:
|
||||
pool_id_list: 삭제할 Pool ID 목록
|
||||
|
||||
Returns:
|
||||
dict: 삭제 결과
|
||||
"""
|
||||
url = f"{self.base_url}/pools"
|
||||
params = {"poolIdList": ",".join(pool_id_list)}
|
||||
|
||||
logger.info(f"Pool 삭제 요청: {pool_id_list}")
|
||||
return self._delete(url, params)
|
||||
|
||||
# ==================== Health Check ====================
|
||||
|
||||
def get_health_check_list(
|
||||
self,
|
||||
health_check_id_list: Optional[List[str]] = None,
|
||||
search_health_check_name: Optional[str] = None,
|
||||
page: int = 1,
|
||||
limit: int = 50,
|
||||
) -> dict:
|
||||
"""
|
||||
Health Check 목록 조회
|
||||
|
||||
Args:
|
||||
health_check_id_list: Health Check ID 목록
|
||||
search_health_check_name: 검색할 Health Check 이름
|
||||
page: 페이지 번호
|
||||
limit: 페이지당 항목 수
|
||||
|
||||
Returns:
|
||||
dict: Health Check 목록
|
||||
"""
|
||||
url = f"{self.base_url}/health-checks"
|
||||
params = {"page": page, "limit": limit}
|
||||
|
||||
if health_check_id_list:
|
||||
params["healthCheckIdList"] = ",".join(health_check_id_list)
|
||||
if search_health_check_name:
|
||||
params["searchHealthCheckName"] = search_health_check_name
|
||||
|
||||
return self._get(url, params)
|
||||
|
||||
def get_health_check(self, health_check_id: str) -> dict:
|
||||
"""
|
||||
Health Check 상세 조회
|
||||
|
||||
Args:
|
||||
health_check_id: Health Check ID
|
||||
|
||||
Returns:
|
||||
dict: Health Check 상세 정보
|
||||
"""
|
||||
url = f"{self.base_url}/health-checks"
|
||||
params = {"healthCheckIdList": health_check_id}
|
||||
result = self._get(url, params)
|
||||
health_checks = result.get("healthCheckList", [])
|
||||
if health_checks:
|
||||
return {"healthCheck": health_checks[0]}
|
||||
raise NHNCloudAPIError(f"Health Check를 찾을 수 없습니다: {health_check_id}", code=404)
|
||||
|
||||
def create_health_check(
|
||||
self,
|
||||
health_check_name: str,
|
||||
health_check_protocol: str,
|
||||
health_check_port: int,
|
||||
health_check_interval: int = 30,
|
||||
health_check_timeout: int = 5,
|
||||
health_check_retries: int = 2,
|
||||
health_check_path: Optional[str] = None,
|
||||
health_check_expected_codes: Optional[str] = None,
|
||||
health_check_expected_body: Optional[str] = None,
|
||||
health_check_request_header_list: Optional[List[dict]] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Health Check 생성
|
||||
|
||||
Args:
|
||||
health_check_name: Health Check 이름
|
||||
health_check_protocol: 프로토콜 (HTTPS, HTTP, TCP)
|
||||
health_check_port: 포트
|
||||
health_check_interval: 체크 간격 (초)
|
||||
health_check_timeout: 타임아웃 (초)
|
||||
health_check_retries: 재시도 횟수
|
||||
health_check_path: 경로 (HTTP/HTTPS 전용)
|
||||
health_check_expected_codes: 예상 응답 코드 (HTTP/HTTPS 전용)
|
||||
health_check_expected_body: 예상 응답 본문 (HTTP/HTTPS 전용)
|
||||
health_check_request_header_list: 요청 헤더 목록 (HTTP/HTTPS 전용)
|
||||
|
||||
Returns:
|
||||
dict: 생성된 Health Check 정보
|
||||
"""
|
||||
url = f"{self.base_url}/health-checks"
|
||||
payload = {
|
||||
"healthCheck": {
|
||||
"healthCheckName": health_check_name,
|
||||
"healthCheckProtocol": health_check_protocol,
|
||||
"healthCheckPort": health_check_port,
|
||||
"healthCheckInterval": health_check_interval,
|
||||
"healthCheckTimeout": health_check_timeout,
|
||||
"healthCheckRetries": health_check_retries,
|
||||
}
|
||||
}
|
||||
|
||||
# HTTP/HTTPS 전용 옵션
|
||||
if health_check_path:
|
||||
payload["healthCheck"]["healthCheckPath"] = health_check_path
|
||||
if health_check_expected_codes:
|
||||
payload["healthCheck"]["healthCheckExpectedCodes"] = health_check_expected_codes
|
||||
if health_check_expected_body:
|
||||
payload["healthCheck"]["healthCheckExpectedBody"] = health_check_expected_body
|
||||
if health_check_request_header_list:
|
||||
payload["healthCheck"]["healthCheckRequestHeaderList"] = health_check_request_header_list
|
||||
|
||||
logger.info(f"Health Check 생성 요청: {health_check_name}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def update_health_check(
|
||||
self,
|
||||
health_check_id: str,
|
||||
health_check_name: Optional[str] = None,
|
||||
health_check_protocol: Optional[str] = None,
|
||||
health_check_port: Optional[int] = None,
|
||||
health_check_interval: Optional[int] = None,
|
||||
health_check_timeout: Optional[int] = None,
|
||||
health_check_retries: Optional[int] = None,
|
||||
health_check_path: Optional[str] = None,
|
||||
health_check_expected_codes: Optional[str] = None,
|
||||
health_check_expected_body: Optional[str] = None,
|
||||
health_check_request_header_list: Optional[List[dict]] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Health Check 수정
|
||||
|
||||
Args:
|
||||
health_check_id: Health Check ID
|
||||
(나머지 파라미터는 create_health_check과 동일)
|
||||
|
||||
Returns:
|
||||
dict: 수정된 Health Check 정보
|
||||
"""
|
||||
url = f"{self.base_url}/health-checks/{health_check_id}"
|
||||
health_check_data = {}
|
||||
|
||||
if health_check_name is not None:
|
||||
health_check_data["healthCheckName"] = health_check_name
|
||||
if health_check_protocol is not None:
|
||||
health_check_data["healthCheckProtocol"] = health_check_protocol
|
||||
if health_check_port is not None:
|
||||
health_check_data["healthCheckPort"] = health_check_port
|
||||
if health_check_interval is not None:
|
||||
health_check_data["healthCheckInterval"] = health_check_interval
|
||||
if health_check_timeout is not None:
|
||||
health_check_data["healthCheckTimeout"] = health_check_timeout
|
||||
if health_check_retries is not None:
|
||||
health_check_data["healthCheckRetries"] = health_check_retries
|
||||
if health_check_path is not None:
|
||||
health_check_data["healthCheckPath"] = health_check_path
|
||||
if health_check_expected_codes is not None:
|
||||
health_check_data["healthCheckExpectedCodes"] = health_check_expected_codes
|
||||
if health_check_expected_body is not None:
|
||||
health_check_data["healthCheckExpectedBody"] = health_check_expected_body
|
||||
if health_check_request_header_list is not None:
|
||||
health_check_data["healthCheckRequestHeaderList"] = health_check_request_header_list
|
||||
|
||||
payload = {"healthCheck": health_check_data}
|
||||
|
||||
logger.info(f"Health Check 수정 요청: {health_check_id}")
|
||||
return self._put(url, payload)
|
||||
|
||||
def delete_health_checks(self, health_check_id_list: List[str]) -> dict:
|
||||
"""
|
||||
Health Check 삭제
|
||||
|
||||
Args:
|
||||
health_check_id_list: 삭제할 Health Check ID 목록
|
||||
|
||||
Returns:
|
||||
dict: 삭제 결과
|
||||
"""
|
||||
url = f"{self.base_url}/health-checks"
|
||||
params = {"healthCheckIdList": ",".join(health_check_id_list)}
|
||||
|
||||
logger.info(f"Health Check 삭제 요청: {health_check_id_list}")
|
||||
return self._delete(url, params)
|
||||
@ -51,6 +51,13 @@ class ApiNks(BaseAPI):
|
||||
headers.update(extra_headers)
|
||||
return headers
|
||||
|
||||
# ==================== Supports ====================
|
||||
|
||||
def get_supports(self) -> dict:
|
||||
"""지원되는 Kubernetes 버전 및 작업 종류 조회"""
|
||||
url = f"{self.nks_url}/v1/supports"
|
||||
return self._get(url)
|
||||
|
||||
# ==================== Cluster ====================
|
||||
|
||||
def get_cluster_list(self) -> dict:
|
||||
@ -195,6 +202,78 @@ class ApiNks(BaseAPI):
|
||||
logger.info(f"Private 클러스터 생성 요청: {cluster_name}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def create_cluster(
|
||||
self,
|
||||
cluster_name: str,
|
||||
vpc_id: str,
|
||||
subnet_id: str,
|
||||
instance_type: str,
|
||||
keypair_name: str,
|
||||
kubernetes_version: str,
|
||||
availability_zone: str,
|
||||
is_public: bool = True,
|
||||
external_network_id: Optional[str] = None,
|
||||
external_subnet_id: Optional[str] = None,
|
||||
node_count: int = 1,
|
||||
boot_volume_size: int = 50,
|
||||
boot_volume_type: str = "General SSD",
|
||||
node_image: Optional[str] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
클러스터 생성 (Public/Private 분기)
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
vpc_id: VPC ID
|
||||
subnet_id: 서브넷 ID
|
||||
instance_type: 인스턴스 타입 (Flavor ID)
|
||||
keypair_name: Keypair 이름
|
||||
kubernetes_version: Kubernetes 버전 (예: v1.28.3)
|
||||
availability_zone: 가용 영역 (예: kr-pub-a)
|
||||
is_public: Public 클러스터 여부 (기본 True)
|
||||
external_network_id: 외부 네트워크 ID (Public 클러스터 필수)
|
||||
external_subnet_id: 외부 서브넷 ID (Public 클러스터 필수)
|
||||
node_count: 노드 수 (기본 1)
|
||||
boot_volume_size: 부팅 볼륨 크기 (GB, 기본 50)
|
||||
boot_volume_type: 볼륨 타입 (기본 "General SSD")
|
||||
node_image: 노드 이미지 ID (기본 Ubuntu 20.04)
|
||||
|
||||
Returns:
|
||||
dict: 생성된 클러스터 정보
|
||||
"""
|
||||
if is_public:
|
||||
if not external_network_id or not external_subnet_id:
|
||||
raise ValueError("Public 클러스터에는 external_network_id와 external_subnet_id가 필요합니다.")
|
||||
return self.create_public_cluster(
|
||||
cluster_name=cluster_name,
|
||||
vpc_id=vpc_id,
|
||||
subnet_id=subnet_id,
|
||||
instance_type=instance_type,
|
||||
keypair_name=keypair_name,
|
||||
kubernetes_version=kubernetes_version,
|
||||
external_network_id=external_network_id,
|
||||
external_subnet_id=external_subnet_id,
|
||||
availability_zone=availability_zone,
|
||||
node_count=node_count,
|
||||
boot_volume_size=boot_volume_size,
|
||||
boot_volume_type=boot_volume_type,
|
||||
node_image=node_image,
|
||||
)
|
||||
else:
|
||||
return self.create_private_cluster(
|
||||
cluster_name=cluster_name,
|
||||
vpc_id=vpc_id,
|
||||
subnet_id=subnet_id,
|
||||
instance_type=instance_type,
|
||||
keypair_name=keypair_name,
|
||||
kubernetes_version=kubernetes_version,
|
||||
availability_zone=availability_zone,
|
||||
node_count=node_count,
|
||||
boot_volume_size=boot_volume_size,
|
||||
boot_volume_type=boot_volume_type,
|
||||
node_image=node_image,
|
||||
)
|
||||
|
||||
def delete_cluster(self, cluster_name: str) -> dict:
|
||||
"""클러스터 삭제"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}"
|
||||
@ -212,3 +291,349 @@ class ApiNks(BaseAPI):
|
||||
"""노드 그룹 상세 조회"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/nodegroups/{nodegroup_name}"
|
||||
return self._get(url)
|
||||
|
||||
def create_nodegroup(
|
||||
self,
|
||||
cluster_name: str,
|
||||
nodegroup_name: str,
|
||||
instance_type: str,
|
||||
node_count: int = 1,
|
||||
availability_zone: Optional[str] = None,
|
||||
boot_volume_size: int = 50,
|
||||
boot_volume_type: str = "General SSD",
|
||||
node_image: Optional[str] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
노드 그룹 생성
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
nodegroup_name: 노드 그룹 이름
|
||||
instance_type: 인스턴스 타입 (Flavor ID)
|
||||
node_count: 노드 수 (기본 1)
|
||||
availability_zone: 가용 영역 (예: kr-pub-a)
|
||||
boot_volume_size: 부팅 볼륨 크기 (GB, 기본 50)
|
||||
boot_volume_type: 볼륨 타입 (기본 "General SSD")
|
||||
node_image: 노드 이미지 ID (기본 Ubuntu 20.04)
|
||||
|
||||
Returns:
|
||||
dict: 생성된 노드 그룹 정보
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/nodegroups"
|
||||
|
||||
payload = {
|
||||
"name": nodegroup_name,
|
||||
"flavor_id": instance_type,
|
||||
"node_count": node_count,
|
||||
"labels": {
|
||||
"boot_volume_size": str(boot_volume_size),
|
||||
"boot_volume_type": boot_volume_type,
|
||||
"node_image": node_image or self.default_node_image,
|
||||
},
|
||||
}
|
||||
|
||||
if availability_zone:
|
||||
payload["labels"]["availability_zone"] = availability_zone
|
||||
|
||||
logger.info(f"노드 그룹 생성 요청: {cluster_name}/{nodegroup_name}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def delete_nodegroup(self, cluster_name: str, nodegroup_name: str) -> dict:
|
||||
"""노드 그룹 삭제"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/nodegroups/{nodegroup_name}"
|
||||
logger.info(f"노드 그룹 삭제 요청: {cluster_name}/{nodegroup_name}")
|
||||
return self._delete(url)
|
||||
|
||||
def start_node(self, cluster_name: str, nodegroup_name: str, node_id: str) -> dict:
|
||||
"""
|
||||
워커 노드 시작
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
nodegroup_name: 노드 그룹 이름
|
||||
node_id: 노드 ID
|
||||
|
||||
Returns:
|
||||
dict: 응답 결과
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/nodegroups/{nodegroup_name}/start_node"
|
||||
payload = {"node_id": node_id}
|
||||
logger.info(f"워커 노드 시작 요청: {cluster_name}/{nodegroup_name}/{node_id}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def stop_node(self, cluster_name: str, nodegroup_name: str, node_id: str) -> dict:
|
||||
"""
|
||||
워커 노드 중지
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
nodegroup_name: 노드 그룹 이름
|
||||
node_id: 노드 ID
|
||||
|
||||
Returns:
|
||||
dict: 응답 결과
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/nodegroups/{nodegroup_name}/stop_node"
|
||||
payload = {"node_id": node_id}
|
||||
logger.info(f"워커 노드 중지 요청: {cluster_name}/{nodegroup_name}/{node_id}")
|
||||
return self._post(url, payload)
|
||||
|
||||
# ==================== Cluster Operations ====================
|
||||
|
||||
def resize_cluster(self, cluster_name: str, node_count: int) -> dict:
|
||||
"""
|
||||
클러스터 노드 수 조정 (리사이즈)
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
node_count: 조정할 노드 수
|
||||
|
||||
Returns:
|
||||
dict: 응답 결과
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/actions/resize"
|
||||
payload = {"node_count": node_count}
|
||||
logger.info(f"클러스터 리사이즈 요청: {cluster_name}, node_count={node_count}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def upgrade_cluster(self, cluster_name: str, kubernetes_version: str) -> dict:
|
||||
"""
|
||||
클러스터 Kubernetes 버전 업그레이드
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
kubernetes_version: 업그레이드할 Kubernetes 버전
|
||||
|
||||
Returns:
|
||||
dict: 응답 결과
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/actions/upgrade"
|
||||
payload = {"kube_tag": kubernetes_version}
|
||||
logger.info(f"클러스터 업그레이드 요청: {cluster_name}, version={kubernetes_version}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def get_cluster_events(self, cluster_uuid: str) -> dict:
|
||||
"""
|
||||
클러스터 작업 이력 목록 조회
|
||||
|
||||
Args:
|
||||
cluster_uuid: 클러스터 UUID
|
||||
|
||||
Returns:
|
||||
dict: 작업 이력 목록
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_uuid}/events"
|
||||
return self._get(url)
|
||||
|
||||
def get_cluster_event(self, cluster_uuid: str, event_uuid: str) -> dict:
|
||||
"""
|
||||
클러스터 작업 이력 상세 조회
|
||||
|
||||
Args:
|
||||
cluster_uuid: 클러스터 UUID
|
||||
event_uuid: 작업 이력 UUID
|
||||
|
||||
Returns:
|
||||
dict: 작업 이력 상세 정보
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_uuid}/events/{event_uuid}"
|
||||
return self._get(url)
|
||||
|
||||
# ==================== Autoscaler ====================
|
||||
|
||||
def get_autoscale_config(self, cluster_name: str, nodegroup_name: str) -> dict:
|
||||
"""
|
||||
오토스케일러 설정 조회
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
nodegroup_name: 노드 그룹 이름
|
||||
|
||||
Returns:
|
||||
dict: 오토스케일러 설정
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/nodegroups/{nodegroup_name}/autoscale"
|
||||
return self._get(url)
|
||||
|
||||
def set_autoscale_config(
|
||||
self,
|
||||
cluster_name: str,
|
||||
nodegroup_name: str,
|
||||
ca_enable: bool,
|
||||
ca_max_node_count: Optional[int] = None,
|
||||
ca_min_node_count: Optional[int] = None,
|
||||
ca_scale_down_enable: Optional[bool] = None,
|
||||
ca_scale_down_delay_after_add: Optional[int] = None,
|
||||
ca_scale_down_unneeded_time: Optional[int] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
오토스케일러 설정 변경
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
nodegroup_name: 노드 그룹 이름
|
||||
ca_enable: 오토스케일러 활성화 여부
|
||||
ca_max_node_count: 최대 노드 수
|
||||
ca_min_node_count: 최소 노드 수
|
||||
ca_scale_down_enable: 스케일 다운 활성화 여부
|
||||
ca_scale_down_delay_after_add: 스케일 다운 지연 시간 (분)
|
||||
ca_scale_down_unneeded_time: 불필요 노드 대기 시간 (분)
|
||||
|
||||
Returns:
|
||||
dict: 변경된 오토스케일러 설정
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/nodegroups/{nodegroup_name}/autoscale"
|
||||
|
||||
payload = {"ca_enable": ca_enable}
|
||||
|
||||
if ca_max_node_count is not None:
|
||||
payload["ca_max_node_count"] = ca_max_node_count
|
||||
if ca_min_node_count is not None:
|
||||
payload["ca_min_node_count"] = ca_min_node_count
|
||||
if ca_scale_down_enable is not None:
|
||||
payload["ca_scale_down_enable"] = ca_scale_down_enable
|
||||
if ca_scale_down_delay_after_add is not None:
|
||||
payload["ca_scale_down_delay_after_add"] = ca_scale_down_delay_after_add
|
||||
if ca_scale_down_unneeded_time is not None:
|
||||
payload["ca_scale_down_unneeded_time"] = ca_scale_down_unneeded_time
|
||||
|
||||
logger.info(f"오토스케일러 설정 변경 요청: {cluster_name}/{nodegroup_name}")
|
||||
return self._post(url, payload)
|
||||
|
||||
# ==================== Node Group Configuration ====================
|
||||
|
||||
def upgrade_nodegroup(
|
||||
self,
|
||||
cluster_name: str,
|
||||
nodegroup_name: str,
|
||||
kubernetes_version: str,
|
||||
max_unavailable_worker: int = 1,
|
||||
) -> dict:
|
||||
"""
|
||||
노드 그룹 Kubernetes 버전 업그레이드
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
nodegroup_name: 노드 그룹 이름
|
||||
kubernetes_version: 업그레이드할 Kubernetes 버전
|
||||
max_unavailable_worker: 동시 업그레이드 가능한 노드 수
|
||||
|
||||
Returns:
|
||||
dict: 응답 결과
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/nodegroups/{nodegroup_name}/upgrade"
|
||||
payload = {
|
||||
"kube_tag": kubernetes_version,
|
||||
"max_unavailable_worker": max_unavailable_worker,
|
||||
}
|
||||
logger.info(f"노드 그룹 업그레이드 요청: {cluster_name}/{nodegroup_name}, version={kubernetes_version}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def set_nodegroup_userscript(
|
||||
self,
|
||||
cluster_name: str,
|
||||
nodegroup_name: str,
|
||||
userscript: str,
|
||||
) -> dict:
|
||||
"""
|
||||
노드 그룹 사용자 스크립트 설정
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
nodegroup_name: 노드 그룹 이름
|
||||
userscript: 사용자 스크립트 내용
|
||||
|
||||
Returns:
|
||||
dict: 응답 결과
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/nodegroups/{nodegroup_name}/userscript"
|
||||
payload = {"userscript": userscript}
|
||||
logger.info(f"노드 그룹 사용자 스크립트 설정 요청: {cluster_name}/{nodegroup_name}")
|
||||
return self._post(url, payload)
|
||||
|
||||
def update_nodegroup(
|
||||
self,
|
||||
cluster_name: str,
|
||||
nodegroup_name: str,
|
||||
instance_type: Optional[str] = None,
|
||||
node_count: Optional[int] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
노드 그룹 설정 변경 (인스턴스 타입, 노드 수 등)
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
nodegroup_name: 노드 그룹 이름
|
||||
instance_type: 인스턴스 타입 (Flavor ID)
|
||||
node_count: 노드 수
|
||||
|
||||
Returns:
|
||||
dict: 변경된 노드 그룹 정보
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/nodegroups/{nodegroup_name}"
|
||||
|
||||
payload = {}
|
||||
if instance_type is not None:
|
||||
payload["flavor_id"] = instance_type
|
||||
if node_count is not None:
|
||||
payload["node_count"] = node_count
|
||||
|
||||
logger.info(f"노드 그룹 설정 변경 요청: {cluster_name}/{nodegroup_name}")
|
||||
return self._patch(url, payload)
|
||||
|
||||
# ==================== Certificates ====================
|
||||
|
||||
def renew_certificates(self, cluster_name: str) -> dict:
|
||||
"""
|
||||
클러스터 인증서 갱신
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
|
||||
Returns:
|
||||
dict: 응답 결과
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/certificates/{cluster_name}"
|
||||
logger.info(f"클러스터 인증서 갱신 요청: {cluster_name}")
|
||||
return self._patch(url, {})
|
||||
|
||||
# ==================== API Endpoint IP ACL ====================
|
||||
|
||||
def get_api_endpoint_ipacl(self, cluster_name: str) -> dict:
|
||||
"""
|
||||
API 엔드포인트 IP 접근 제어 조회
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
|
||||
Returns:
|
||||
dict: IP 접근 제어 설정
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/api_ep_ipacl"
|
||||
return self._get(url)
|
||||
|
||||
def set_api_endpoint_ipacl(
|
||||
self,
|
||||
cluster_name: str,
|
||||
enable: bool,
|
||||
allowed_cidrs: Optional[list] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
API 엔드포인트 IP 접근 제어 설정
|
||||
|
||||
Args:
|
||||
cluster_name: 클러스터 이름
|
||||
enable: IP 접근 제어 활성화 여부
|
||||
allowed_cidrs: 허용할 CIDR 목록 (예: ["192.168.0.0/24", "10.0.0.0/8"])
|
||||
|
||||
Returns:
|
||||
dict: 변경된 IP 접근 제어 설정
|
||||
"""
|
||||
url = f"{self.nks_url}/v1/clusters/{cluster_name}/api_ep_ipacl"
|
||||
|
||||
payload = {"enable": enable}
|
||||
if allowed_cidrs is not None:
|
||||
payload["allowed_cidrs"] = allowed_cidrs
|
||||
|
||||
logger.info(f"API 엔드포인트 IP 접근 제어 설정 요청: {cluster_name}")
|
||||
return self._post(url, payload)
|
||||
|
||||
Reference in New Issue
Block a user