API를 활용한 네이버클라우드 Target group 조회/생성/적용

[사전 필수 요소]

• API 키 발급
  • Sub account 계정도 API키 발급이 가능합니다.

---

[API 조회 코드(Python)]

import requests
import time
import hashlib
import hmac
import base64

# 액세스 키와 시크릿 키 설정
access_key = 'access_key'   # 실제 액세스 키 ID
secret_key = 'secret_key'      # 실제 시크릿 키

# 요청 정보 설정
http_method = "GET"
host = "ncloud.apigw.ntruss.com"
url = "/vloadbalancer/v2/getTargetGroupList" ## 타겟그룹리스트 확인
timestamp = str(int(time.time() * 1000))  # 현재 시간의 타임스탬프 (밀리초 단위)

# StringToSign 생성 (호스트 URL 제거)
message = "{} {}\n{}\n{}".format(http_method, url, timestamp, access_key)

# 시그니처 생성
signingKey = bytes(secret_key, 'UTF-8')
message = bytes(message, 'UTF-8')
signature = base64.b64encode(hmac.new(signingKey, message, digestmod=hashlib.sha256).digest())

# 헤더 설정
headers = {
    'x-ncp-apigw-timestamp': timestamp,
    'x-ncp-iam-access-key': access_key,
    'x-ncp-apigw-signature-v2': signature.decode('UTF-8')
}

# API 호출
url_full = "https://{}{}".format(host, url)
response = requests.get(url_full, headers=headers)

# 응답 출력
print(response.status_code)
print(response.text)

 

---

[결과 조회]

 

---

[API를 통한 Target 생성 코드(Python)]

 

import requests
import time
import hashlib
import hmac
import base64
import urllib.parse
import json

# 액세스 키와 시크릿 키 설정
access_key = 'access_key'    # 실제 액세스 키
secret_key = 'secret_key'       # 실제 시크릿 키

# 요청 정보 설정
http_method = "POST"
host = "ncloud.apigw.ntruss.com"
base_url = "/vloadbalancer/v2/createTargetGroup"
timestamp = str(int(time.time() * 1000))  # 현재 시간의 타임스탬프 (밀리초 단위)

# 파라미터 설정 (모든 파라미터를 쿼리 스트링으로 이동)
params = {
    "regionCode": "KR",
    "vpcNo": "*****",                        # 실제 VPC 
    "targetGroupName": "test-api-target",
    "targetTypeCode": "VSVR",
    "targetGroupProtocolTypeCode": "PROXY_TCP",
    "targetGroupPort": "5500",
    "healthCheckProtocolTypeCode": "TCP",
    "healthCheckPort": "80",
    "healthCheckCycle": "30",
    "healthCheckTimeout": "5",
    "healthCheckUpThreshold": "3",
    "healthCheckDownThreshold": "3",
    "useProxyProtocol": "false",
    "useStickySession": "false",
    "targetGroupDescription": "api-test타겟그룹",
    "responseFormatType": "json"
}

# 파라미터 정렬
sorted_params = dict(sorted(params.items()))

# 쿼리 스트링 생성
query_string = urllib.parse.urlencode(sorted_params)
url_with_query = base_url + "?" + query_string

# StringToSign 생성 (정렬된 쿼리 스트링 포함)
message = "{} {}\n{}\n{}".format(http_method, url_with_query, timestamp, access_key)

# 시그니처 생성
signingKey = bytes(secret_key, 'UTF-8')
message_bytes = bytes(message, 'UTF-8')
signature = base64.b64encode(hmac.new(signingKey, message_bytes, digestmod=hashlib.sha256).digest())
signature_str = signature.decode('UTF-8')

# 헤더 설정
headers = {
    'x-ncp-apigw-timestamp': timestamp,
    'x-ncp-iam-access-key': access_key,
    'x-ncp-apigw-signature-v2': signature_str
}

# API 호출 (요청 본문 없음)
url_full = "https://{}{}".format(host, url_with_query)
response = requests.post(url_full, headers=headers)

# 응답 출력
if response.status_code == 200:
    data = response.json()
    print("타겟 그룹 생성 성공:")
    print(json.dumps(data, indent=4, ensure_ascii=False))
else:
    print("타겟 그룹 생성 실패:")
    print("Status Code:", response.status_code)
    try:
        error_data = response.json()
        print(json.dumps(error_data, indent=4, ensure_ascii=False))
    except json.JSONDecodeError:
        print(response.text)

---

[API로 생성한 타겟그룹]

---

[API로 여러 개의 타겟그룹 동시 생성 코드(Python)]

import requests
import time
import hashlib
import hmac
import base64
import urllib.parse
import json
import os

# 액세스 키와 시크릿 키 설정 (환경 변수 사용 권장)
access_key = 'access_key'  # 실제 액세스 키 ID로 대체하세요
secret_key = 'secret_key'  # 실제 시크릿 키로 대체하세요

# 요청 정보 설정
http_method = "POST"
host = "ncloud.apigw.ntruss.com"
base_url = "/vloadbalancer/v2/createTargetGroup"

# 공통 파라미터 설정
common_params = {
    "regionCode": "KR",
    "vpcNo": "74794",  # 실제 VPC 번호로 대체하세요
    "targetTypeCode": "VSVR",
    "targetGroupProtocolTypeCode": "PROXY_TCP",
    "healthCheckProtocolTypeCode": "TCP",
    "healthCheckPort": "80",
    "healthCheckCycle": "30",
    "healthCheckTimeout": "5",
    "healthCheckUpThreshold": "3",
    "healthCheckDownThreshold": "3",
    "useProxyProtocol": "false",
    "useStickySession": "false",
    "targetGroupDescription": "api-test타겟그룹",
    "responseFormatType": "json"
}

# 포트 번호 범위 설정
start_port = 6500
end_port = 6550

for port in range(start_port, end_port + 1):
    # 각 타겟 그룹에 대한 파라미터 설정
    params = common_params.copy()
    params["targetGroupName"] = f"example-tg-{port}"
    params["targetGroupPort"] = str(port)  # 포트 번호를 문자열로 변환
    # 타임스탬프 업데이트
    timestamp = str(int(time.time() * 1000))

    # 파라미터 정렬
    sorted_params = dict(sorted(params.items()))

    # 쿼리 스트링 생성
    query_string = urllib.parse.urlencode(sorted_params)
    url_with_query = base_url + "?" + query_string

    # StringToSign 생성 (정렬된 쿼리 스트링 포함)
    message = "{} {}\n{}\n{}".format(http_method, url_with_query, timestamp, access_key)

    # 시그니처 생성
    signingKey = bytes(secret_key, 'UTF-8')
    message_bytes = bytes(message, 'UTF-8')
    signature = base64.b64encode(hmac.new(signingKey, message_bytes, digestmod=hashlib.sha256).digest())
    signature_str = signature.decode('UTF-8')

    # 헤더 설정
    headers = {
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key,
        'x-ncp-apigw-signature-v2': signature_str
    }

    # API 호출 (요청 본문 없음)
    url_full = "https://{}{}".format(host, url_with_query)
    response = requests.post(url_full, headers=headers)

    # 응답 출력
    if response.status_code == 200:
        data = response.json()
        print(f"타겟 그룹 생성 성공: port{port}")
    else:
        print(f"타겟 그룹 생성 실패: port{port}")
        print("Status Code:", response.status_code)
        try:
            error_data = response.json()
            print(json.dumps(error_data, indent=4, ensure_ascii=False))
        except json.JSONDecodeError:
            print(response.text)

    # API 호출 간 딜레이 추가 (선택 사항)
    time.sleep(2.0)  # 1.0초 대기 (필요에 따라 조정)

 

---

[API로 타겟그룹 적용 코드(Python)]

 

import requests
import time
import hashlib
import hmac
import base64
import urllib.parse
import json

# 액세스 키와 시크릿 키 설정 (직접 입력)
access_key = 'access_key  # 실제 액세스 키 ID로 대체하세요
secret_key = 'secret_key'  # 실제 시크릿 키로 대체하세요

# 타겟 그룹 번호 조회 함수
def get_target_group_no(target_group_name, vpc_no):
    http_method = "GET"
    host = "ncloud.apigw.ntruss.com"  # 타겟 그룹 조회 호스트
    base_url = "/vloadbalancer/v2/getTargetGroupList"
    timestamp = str(int(time.time() * 1000))

    params = {
        "vpcNo": vpc_no,
        "responseFormatType": "json"
    }
    sorted_params = dict(sorted(params.items()))
    query_string = urllib.parse.urlencode(sorted_params)
    url_with_query = f"{base_url}?{query_string}"

    # StringToSign 생성
    message = f"{http_method} {url_with_query}\n{timestamp}\n{access_key}"
    signing_key = bytes(secret_key, 'UTF-8')
    message_bytes = bytes(message, 'UTF-8')

    signature = base64.b64encode(hmac.new(signing_key, message_bytes, digestmod=hashlib.sha256).digest())
    signature_str = signature.decode('UTF-8')

    # 헤더 설정
    headers = {
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key,
        'x-ncp-apigw-signature-v2': signature_str
    }

    url_full = f"https://{host}{url_with_query}"
    response = requests.get(url_full, headers=headers)

    if response.status_code == 200:
        data = response.json()
        target_groups = data.get('getTargetGroupListResponse', {}).get('targetGroupList', [])
        for tg in target_groups:
            if tg.get('targetGroupName') == target_group_name:
                return tg.get('targetGroupNo')
    else:
        print(f"타겟 그룹 조회 실패: {target_group_name}")
        print("Status Code:", response.status_code)
        print(response.text)

    return None

# 타겟 그룹에 타겟 설정 함수
def set_target_group_instance(region_code, target_group_no, target_no):
    http_method = "POST"
    host = "ncloud.apigw.ntruss.com"  # 타겟 그룹 설정 호스트
    base_url = "/vloadbalancer/v2/setTarget"
    timestamp = str(int(time.time() * 1000))

    # 쿼리 매개변수 설정
    params = {
        "regionCode": region_code,
        "targetGroupNo": target_group_no,
        "targetNoList.1": target_no
    }
    sorted_params = dict(sorted(params.items()))
    query_string = urllib.parse.urlencode(sorted_params)
    url_with_query = f"{base_url}?{query_string}"

    # StringToSign 생성
    message = f"{http_method} {url_with_query}\n{timestamp}\n{access_key}"
    signing_key = bytes(secret_key, 'UTF-8')
    message_bytes = bytes(message, 'UTF-8')

    signature = base64.b64encode(hmac.new(signing_key, message_bytes, digestmod=hashlib.sha256).digest())
    signature_str = signature.decode('UTF-8')

    # 헤더 설정
    headers = {
        'Content-Type': 'application/json',  # JSON 형식임을 명시
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key,
        'x-ncp-apigw-signature-v2': signature_str
    }

    url_full = f"https://{host}{url_with_query}"

    # 디버깅용 출력 (선택 사항)
    print(f"요청 URL: {url_full}")
    print(f"요청 본문: {{}}")  # setTarget API는 쿼리 매개변수를 사용하므로 본문은 비워둡니다.

    # 요청 본문을 비워둡니다.
    response = requests.post(url_full, headers=headers, data={})

    if response.status_code == 200:
        print(f"타겟 그룹에 타겟 설정 성공: Target Group No {target_group_no}")
    else:
        print(f"타겟 그룹에 타겟 설정 실패: Target Group No {target_group_no}")
        print("Status Code:", response.status_code)
        print(response.text)

# 메인 로직
if __name__ == "__main__":
    # VPC 번호 설정
    vpc_no = "74794"  # 실제 VPC 번호로 대체하세요

    # 서버 인스턴스 번호 설정 (예시)
    server_instance_no = "26360985"  # 실제 서버 인스턴스 번호로 대체하세요

    # 리전 코드 설정 (예시: KR)
    region_code = "KR"  # 실제 리전 코드로 대체하세요

    # 포트 번호 범위 설정
    start_port = 6500
    end_port = 6550

    for port in range(start_port, end_port + 1):
        target_group_name = f"tiger-tg-{port}"  # 타겟 그룹 이름이 정확한지 확인하세요

        # 타겟 그룹 번호 조회
        target_group_no = get_target_group_no(target_group_name, vpc_no)
        if target_group_no is None:
            print(f"타겟 그룹 번호를 찾을 수 없습니다: {target_group_name}")
            continue

        # 타겟 그룹에 타겟 설정
        set_target_group_instance(region_code, target_group_no, server_instance_no)

        # API 호출 간 딜레이 추가 (선택 사항)
        time.sleep(0.5)

 

---

[API LB 리스너에 포트 및 타겟그룹 매핑하는 코드(Python)

import requests
import time
import hashlib
import hmac
import base64
import json

# 액세스 키와 시크릿 키 설정
access_key = 'access_key'
secret_key = 'secret_key'

# 요청 정보 설정
http_method = "POST"
host = "ncloud.apigw.ntruss.com"
base_url = "/vloadbalancer/v2/createLoadBalancerListener"
start_port = 6500
end_port = 6550
start_target_group_no = 2649865  # 시작 타겟 그룹 번호
max_retries = 3  # 최대 재시도 횟수
load_balancer_instance_no = "26736195"  # 로드밸런서 인스턴스 번호


# 로드 밸런서 상태 체크를 위한 함수
def check_load_balancer_status(instance_no):
    status_url = f"https://{host}/vloadbalancer/v2/getLoadBalancerInstance"
    timestamp = str(int(time.time() * 1000))
    params = {
        "loadBalancerInstanceNo": instance_no,
        "responseFormatType": "json"
    }
    message = "{} {}\n{}\n{}".format("GET", "/vloadbalancer/v2/getLoadBalancerInstance", timestamp, access_key)
    signingKey = bytes(secret_key, 'UTF-8')
    message_bytes = bytes(message, 'UTF-8')
    signature = base64.b64encode(hmac.new(signingKey, message_bytes, digestmod=hashlib.sha256).digest())
    signature_str = signature.decode('UTF-8')

    headers = {
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key,
        'x-ncp-apigw-signature-v2': signature_str,
        'Content-Type': 'application/x-www-form-urlencoded'
    }

    # URL 출력
    print(f"로드 밸런서 상태 체크 URL: {status_url}, 파라미터: {params}")

    response = requests.get(status_url, headers=headers, params=params)

    if response.status_code == 200:
        data = response.json()
        print(f"로드 밸런서 상태: {json.dumps(data, indent=4, ensure_ascii=False)}")
    else:
        print(f"로드 밸런서 상태 체크 실패: {response.status_code}")
        print(response.text)


# 로드 밸런서 상태 체크
check_load_balancer_status(load_balancer_instance_no)

# 포트 번호를 올바르게 생성하기 위한 루프
for port in range(start_port, end_port + 1):
    target_group_no = start_target_group_no + (port - start_port)

    retries = 0
    success = False  # 리스너 생성 성공 여부

    while retries < max_retries:
        timestamp = str(int(time.time() * 1000))  # 현재 시간의 타임스탬프 (밀리초 단위)

        params = {
            "regionCode": "KR",
            "loadBalancerInstanceNo": load_balancer_instance_no,
            "protocolTypeCode": "TCP",
            "port": str(port),
            "targetGroupNo": str(target_group_no),
            "responseFormatType": "json"
        }

        message = "{} {}\n{}\n{}".format(http_method, base_url, timestamp, access_key)
        signingKey = bytes(secret_key, 'UTF-8')
        message_bytes = bytes(message, 'UTF-8')
        signature = base64.b64encode(hmac.new(signingKey, message_bytes, digestmod=hashlib.sha256).digest())
        signature_str = signature.decode('UTF-8')

        headers = {
            'x-ncp-apigw-timestamp': timestamp,
            'x-ncp-iam-access-key': access_key,
            'x-ncp-apigw-signature-v2': signature_str,
            'Content-Type': 'application/x-www-form-urlencoded'
        }

        url_full = "https://{}{}".format(host, base_url)

        # URL 출력
        print(f"\n리스너 생성 URL: {url_full}, 파라미터: {params}")

        response = requests.post(url_full, headers=headers, data=params)

        print(f"\n리스너 포트 {port}, 타겟 그룹 {target_group_no}에 대한 응답 상태 코드:", response.status_code)

        if response.status_code == 200:
            try:
                data = response.json()
                print(f"\n포트 {port}의 리스너 생성 성공:")
                print(json.dumps(data, indent=4, ensure_ascii=False))
                success = True  # 성공 플래그 설정
                time.sleep(50)  # 리스너 생성 성공 시 50초 대기
            except json.JSONDecodeError:
                print(f"\n포트 {port}의 리스너 생성 성공, 그러나 JSON 파싱에 실패했습니다.")
                print(response.text)
                success = False  # 성공했지만 파싱 실패
            break  # 성공 시 루프 종료
        else:
            print(f"\n포트 {port}의 리스너 생성 실패:")
            print(f"Status Code: {response.status_code}")
            try:
                error_data = response.json()
                print(json.dumps(error_data, indent=4, ensure_ascii=False))
            except json.JSONDecodeError:
                print(response.text)

            if response.status_code == 401 and "Expired timestamp" in error_data.get("message", ""):
                retries += 1
                print(f"재시도 중... ({retries}/{max_retries})")
                time.sleep(10)  # 재시도 간 대기 시간
            elif response.status_code == 400:
                print("로드 밸런서 인스턴스가 사용 중이지 않거나 잘못된 인스턴스 번호입니다. 확인이 필요합니다.")
                break  # 상태 코드 400 발생 시 루프 종료
            else:
                print("기타 오류 발생, 재시도 중...")
                retries += 1
                time.sleep(10)  # 대기 후 재시도

    if not success:
        print(f"\n포트 {port}의 리스너 생성 실패, 다음 포트로 넘어갑니다.")
        time.sleep(20)  # 포트 변경 전 대기 시간 추가

# 각 API 호출 간 지연 추가
time.sleep(70.0)  # 각 호출 후 50초 대기

---

[결론]

 

API가이드만 충실히 따르면 실행 잘 되네요. 상당히 쉽게 만들어놨습니다.
API 에러코드 또한 직관적으로 알 수 있어서 활용하기 좋습니다.

https://api.ncloud-docs.com/docs/api-overview