""" 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 # ==================== 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 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)