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