Files
icurfer 57526a0f13
All checks were successful
Build And Test / build-and-push (push) Successful in 36s
v0.0.8 | CORS 설정에 X-NHN-Appkey 헤더 허용 추가
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-15 00:32:00 +09:00

640 lines
22 KiB
Python

"""
NHN Cloud NKS (Kubernetes Service) API Module
Kubernetes 클러스터 관리
"""
import logging
from typing import Optional
from dataclasses import dataclass
from .base import BaseAPI, NHNCloudEndpoints, Region
logger = logging.getLogger(__name__)
@dataclass
class NKSNodeImages:
"""NKS 노드 이미지 ID"""
# Ubuntu 20.04 이미지 ID (리전별)
UBUNTU_20_04_KR1 = "1213d033-bdf6-4d73-9763-4e8e57c745fb"
UBUNTU_20_04_KR2 = "dabb6d10-937d-4952-9ce0-1e576e9164e8"
class ApiNks(BaseAPI):
"""NHN Cloud NKS API 클래스"""
def __init__(self, region: str, token: str):
"""
Args:
region: 리전 (kr1: 판교, kr2: 평촌)
token: API 인증 토큰
"""
super().__init__(region, token)
if self.region == Region.KR1:
self.nks_url = NHNCloudEndpoints.NKS_KR1
self.default_node_image = NKSNodeImages.UBUNTU_20_04_KR1
else:
self.nks_url = NHNCloudEndpoints.NKS_KR2
self.default_node_image = NKSNodeImages.UBUNTU_20_04_KR2
def _get_headers(self, extra_headers: Optional[dict] = None) -> dict:
"""NKS API 전용 헤더"""
headers = {
"X-Auth-Token": self.token,
"Accept": "application/json",
"Content-Type": "application/json",
"OpenStack-API-Version": "container-infra latest",
}
if extra_headers:
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:
"""클러스터 목록 조회"""
url = f"{self.nks_url}/v1/clusters"
return self._get(url)
def get_cluster_info(self, cluster_name: str) -> dict:
"""클러스터 상세 조회"""
url = f"{self.nks_url}/v1/clusters/{cluster_name}"
return self._get(url)
def get_cluster_config(self, cluster_name: str) -> str:
"""클러스터 kubeconfig 조회"""
url = f"{self.nks_url}/v1/clusters/{cluster_name}/config"
data = self._get(url)
return data.get("config", "")
def create_public_cluster(
self,
cluster_name: str,
vpc_id: str,
subnet_id: str,
instance_type: str,
keypair_name: str,
kubernetes_version: str,
external_network_id: str,
external_subnet_id: str,
availability_zone: str,
node_count: int = 1,
boot_volume_size: int = 50,
boot_volume_type: str = "General SSD",
node_image: Optional[str] = None,
) -> dict:
"""
Public 클러스터 생성 (외부 접근 가능)
Args:
cluster_name: 클러스터 이름
vpc_id: VPC ID
subnet_id: 서브넷 ID
instance_type: 인스턴스 타입 (Flavor ID)
keypair_name: Keypair 이름
kubernetes_version: Kubernetes 버전 (예: v1.28.3)
external_network_id: 외부 네트워크 ID
external_subnet_id: 외부 서브넷 ID
availability_zone: 가용 영역 (예: kr-pub-a)
node_count: 노드 수 (기본 1)
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"
payload = {
"cluster_template_id": "iaas_console",
"create_timeout": 60,
"fixed_network": vpc_id,
"fixed_subnet": subnet_id,
"flavor_id": instance_type,
"keypair": keypair_name,
"labels": {
"availability_zone": availability_zone,
"boot_volume_size": str(boot_volume_size),
"boot_volume_type": boot_volume_type,
"ca_enable": "false",
"cert_manager_api": "True",
"clusterautoscale": "nodegroupfeature",
"external_network_id": external_network_id,
"external_subnet_id_list": external_subnet_id,
"kube_tag": kubernetes_version,
"master_lb_floating_ip_enabled": "True",
"node_image": node_image or self.default_node_image,
},
"name": cluster_name,
"node_count": str(node_count),
}
logger.info(f"Public 클러스터 생성 요청: {cluster_name}")
return self._post(url, payload)
def create_private_cluster(
self,
cluster_name: str,
vpc_id: str,
subnet_id: str,
instance_type: str,
keypair_name: str,
kubernetes_version: str,
availability_zone: str,
node_count: int = 1,
boot_volume_size: int = 50,
boot_volume_type: str = "General SSD",
node_image: Optional[str] = None,
) -> dict:
"""
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)
node_count: 노드 수 (기본 1)
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"
payload = {
"cluster_template_id": "iaas_console",
"create_timeout": 60,
"fixed_network": vpc_id,
"fixed_subnet": subnet_id,
"flavor_id": instance_type,
"keypair": keypair_name,
"labels": {
"availability_zone": availability_zone,
"boot_volume_size": str(boot_volume_size),
"boot_volume_type": boot_volume_type,
"ca_enable": "false",
"cert_manager_api": "True",
"clusterautoscale": "nodegroupfeature",
"kube_tag": kubernetes_version,
"master_lb_floating_ip_enabled": "False",
"node_image": node_image or self.default_node_image,
},
"name": cluster_name,
"node_count": str(node_count),
}
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}"
logger.info(f"클러스터 삭제 요청: {cluster_name}")
return self._delete(url)
# ==================== Node Group ====================
def get_nodegroup_list(self, cluster_name: str) -> dict:
"""노드 그룹 목록 조회"""
url = f"{self.nks_url}/v1/clusters/{cluster_name}/nodegroups"
return self._get(url)
def get_nodegroup_info(self, cluster_name: str, nodegroup_name: str) -> dict:
"""노드 그룹 상세 조회"""
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)