Cloud/GCP

GCP VPN과 PSC를 통한 Vertex AI (Gemini-API) Private 연결 가이드

달빛궁전- 2025. 8. 12. 10:30
배경 : 온프레미스 데이터센터, AWS, Azure의 서버들이 인터넷을 거치지 않고, VPN을 통해 GCP의 Vertex AI(Gemini) 서비스와 안전하게 통신하는 환경을 구축하는 기술 가이드입니다.
Cloud VPN으로 네트워크를 연결하고 Private Service Connect(PSC)를 이용해 비공개로 서비스를 호출하는 전체 구성 과정을 안내합니다.

목적 : 이 구성을 통해 얻을 수 있는 이점은 다음과 같습니다.

강력한 보안: 민감한 AI 트래픽을 공개 인터넷으로부터 완벽하게 격리합니다.
안정적인 성능: Google 내부 네트워크를 이용해 API에 접속하므로, 일관된 성능과 낮은 지연 시간을 확보할 수 있습니다.
간결한 아키텍처: 복잡한 프록시나 NAT 게이트웨이 없이 온프레미스/타 퍼블릭 클라우드에서 Google 내부 네트워크를 통해 Vertex AI와 안전하게 연동합니다.

본 문서에서는 Vertex AI gemini api 사용시 us-central1-aiplatform.googleapis.com 즉 us-central1리전을 사용했습니다. 사용사유는 한국리전(asia-northeast3)도 gemini api를 지원하기는 하나, 최신 gemini 버전은 아시아에서 지원되지 않는 경우가 많기 때문입니다. AI API경우에는 가용성을 높일 수 있는 global을 사용하는 것을 추천합니다.

 

아키텍처

구성 절차

절차 1: VPC 네트워크 및 서브넷 생성

먼저, API 트래픽이 안전하게 지나갈 GCP 내부의 전용 도로(VPC)를 만듭니다.
  • 메뉴: VPC 네트워크
  • 작업: + VPC 네트워크 만들기
  • 핵심 설정:
    • VPC 이름:gemini-psc-vpc
    • 서브넷 모드: 커스텀
    • 서브넷 이름: onprem-subnet-gemini-an3
    • 리전: asia-northeast3 (서울)
    • IP 범위: 10.10.3.0/24 (AWS, Azure, 온프레미스 네트워크와 겹치지 않게)

절차 2: Cloud VPN 연결 (HA VPN)

온프레미스 방화벽 장비와 GCP VPC를 암호화된 터널로 연결합니다.
Cloud VPN(HA 모드)로 온프레미스-클라우드 간 BGP 기반 암호화 터널을 구성해야 하며, 반드시 라우팅 테이블 충돌에 유의해야 합니다
해당 방안은 별도 포스팅하였으니 해당 문서를 참조 부탁드립니다.

간략한 설명은 아래와 같습니다.
  • 메뉴: 하이브리드 연결 > VPN
  • 작업: + VPN 연결 만들기 > HA VPN
  • 생성할 리소스:
    1. Cloud VPN 게이트웨이: GCP 측의 관문 (asia-northeast3 리전 선택)
    2. 피어 VPN 게이트웨이: 온프레미스 방화벽의 공인 IP를 등록
    3. Cloud Router: 경로 정보를 교환 (BGP)
    4. VPN 터널: 온프레미스와 협의된 공유 보안 비밀(Pre-shared key)을 입력

 

절차 3: Private Service Connect(PSC) 엔드포인트 생성

VPC 내부에 모든 Google API로 통하는 비공개 통로를 만듭니다.
  • 메뉴: VPC 네트워크 > Private Service Connect
  • 작업: + 엔드포인트 연결
  • 설정:
    • 대상: Google API
    • API 번들: 모든 Google API
    • 네트워크 / 서브넷: 1단계에서 만든 vpc  선택
      이때 신규 IP는 해당 VPC-subnet에 속하지 않은 IP로 해야됩니다.
  • 결과: 생성이 완료되면 엔드포인트의 사설 IP 주소를 복사해 둡니다.

 

절차 4: 비공개 DNS 영역 설정

vertex AI gemiani API를 호출하기 위해 *.region-aiplatform-aiplatform 주소를 PSC의 사설 IP로 연결해주는 설정입니다. 
DNS연결 작업을 하지 않는다면, PSC IP주소로만 호출해야 합니다.
  • 메뉴: 네트워크 서비스 > Cloud DNS
  • 작업: + 영역 만들기
  • 핵심 설정:
    • 유형: 비공개
    • Zone name : 알기 쉬운 이름으로 설정
    • DNS 이름:googleapis.com 
    • 네트워크:  선택
  • 레코드 세트 추가: 생성된 영역에 아래 2개의 레코드를 추가합니다.
DNS 이름 유형 데이터 (IPv4 주소 / 정식 이름)
us-central1-aiplatform 
A
3단계에서 복사한 PSC 엔드포인트의 사설 IP
*.us-central1-aiplatform-aiplatform
CNAME
us-central1-aiplatform.googleapis.com.
 
💡 중요: 온프레미스 DNS 사용 시
대부분의 기업 환경처럼 온프레미스에 자체 DNS 서버를 운영하는 경우, 반드시 추가 작업이 필요합니다. 온프레미스, 타CSP DNS 서버에  us-central1-aiplatform.googleapis.com.
도메인에 대한 모든 요청을 GCP로 전달(Conditional Forwarding)하는 규칙을 설정해야 합니다.
이 설정이 없으면 온프레미스 서버가 외부 DNS에 질의하여 공개 IP를 받게 되므로, PSC를 구성한 의미가 사라집니다.

 

이제 외부에서 DNS를 질의할 때 어디에 질의해야될지를 설정해 주어야할 차례입니다.
즉 DNS의 IP 설정이 필요합니다.
DNS Server Policies로 이동하여 “Create Policy” 선택하여 새로운 DNS 정책을 만들어 줍니다.

  • 설정 : 
    • Inbound query forwarding : On
    • DNS64 : off
    • logs : off (실 운영환경에서는 로그가 필요한 상황이라면 on합니다. 로그가 생성되면 상황과 추적이 가능하지만 비용이 증가하게 됩니다.)
    • 네트워크 : 선택


생성 후 다시 DNS정책화면으로 들어오면 “In use by” 확인할 수 있습니다.
해당 IP가 us-central1-aiplatform.googleapis.com 에 대한 DNS 서버이므로 별도로 기록해 둡니다.

 

절차 5: 온프레미스에서 연결 테스트 

모든 구성이 완료된 후, 온프레미스 서버에서 최종 확인을 합니다.
  1. DNS 확인: nslookup 명령어로 주소가 사설 IP로 변환되는지 확인합니다.
    nslookup asia-northeast3-aiplatform.googleapis.com
     
  2. API 호출 테스트(1): 먼저 도메인 주소가 아닌 PSC IP주소로 하는 Python 코드로 Gemini API가 정상 동작하는지 확인합니다.
    import json
import subprocess
import requests
import warnings
from urllib3.exceptions import InsecureRequestWarning

# --- 설정값 ---
# 1. PSC 엔드포인트 IP 주소
PSC_ENDPOINT_IP = "10.10.50.50"

# 2. Google Cloud 프로젝트 ID
PROJECT_ID = "ctu-gcp-dsa2-unit"

# 3. Vertex AI를 사용할 리전
LOCATION = "us-central1"

# 4. 테스트할 모델 ID
MODEL_ID = "gemini-2.0-flash-001"


def get_access_token():
    """gcloud CLI를 통해 Vertex AI용 Access Token을 가져옵니다."""
    try:
        token = subprocess.check_output(
            ["gcloud", "auth", "print-access-token"], 
            text=True
        ).strip()
        print("✅ Access Token을 성공적으로 가져왔습니다. (Vertex AI 인증용)")
        return token
    except (subprocess.CalledProcessError, FileNotFoundError) as e:
        print(f"❌ Access Token을 가져오는 데 실패했습니다: {e}")
        return None

def ask_gemini_via_psc(prompt: str):
    """PSC를 통해 Vertex AI Gemini API를 직접 호출합니다."""

    # 1. 인증: API 키가 아닌, VM의 서비스 계정 권한으로 Access Token을 받습니다.
    token = get_access_token()
    if not token:
        return

    # 2. 주소 설정: 실제 요청은 PSC의 IP 주소로 보냅니다.
    api_endpoint = f"https://{PSC_ENDPOINT_IP}/v1/projects/{PROJECT_ID}/locations/{LOCATION}/publishers/google/models/{MODEL_ID}:generateContent"

    # 3. 헤더 설정: PSC가 요청을 올바르게 전달하도록 설정합니다.
    headers = {
        "Authorization": f"Bearer {token}",          # 인증 토큰
        "Content-Type": "application/json",
        "Host": f"{LOCATION}-aiplatform.googleapis.com" # 목적지(Vertex AI) 명시
    }

    # 4. 요청 본문(payload) 생성 
    #    Gemini API 형식에 맞게 "role": "user"를 추가합니다.
    request_body = {
        "contents": [
            {
                "role": "user", 
                "parts": [
                    {"text": prompt}
                ]
            }
        ]
    }

    print(f"\nPSC 엔드포인트로 요청을 보냅니다: https://{PSC_ENDPOINT_IP}/...")
    
    # SSL 경고를 잠시 비활성화합니다.
    warnings.filterwarnings('ignore', category=InsecureRequestWarning)
    
    try:
        # 5. PSC로 요청 전송 (SSL 인증서 검증은 비활성화)
        response = requests.post(api_endpoint, headers=headers, json=request_body, verify=False, timeout=20)
        response.raise_for_status()

        response_json = response.json()
        text_content = response_json["candidates"][0]["content"]["parts"][0]["text"]

        print("\n--- Gemini 응답 ---")
        print(text_content)
        print("--------------------")
        print("\n✅ PSC를 통한 호출 성공!")

    except requests.exceptions.HTTPError as e:
        print(f"\n❌ API 요청 중 HTTP 오류가 발생했습니다: {e.response.status_code}")
        print("   오류 내용:")
        print(json.dumps(e.response.json(), indent=2))
        
    except Exception as e:
        print(f"\n❌ 예상치 못한 오류가 발생했습니다: {e}")
    finally:
        # 경고 설정을 원래대로 복원합니다.
        warnings.resetwarnings()


if __name__ == "__main__":
    prompt_to_send = "자기소개 부탁해"
    ask_gemini_via_psc(prompt_to_send)
  3. API 호출 테스트(2): 도메인 주소(us-central1-aiplatform.googleapis.com) 테스트하는 Python 코드로 Gemini API가 정상 동작하는지 확인합니다.
    import json
import subprocess
import requests
import warnings
from urllib3.exceptions import InsecureRequestWarning

# --- 설정값 ---
# 1. 도메인 (PSC DNS CNAME이 redirect하는 Vertex AI 엔드포인트)
VERTEX_API_DOMAIN = "us-central1-aiplatform.googleapis.com"
# 2. Google Cloud 프로젝트 ID
PROJECT_ID = "ctu-gcp-dsa2-unit"
# 3. Vertex AI를 사용할 리전
LOCATION = "us-central1"
# 4. 테스트할 모델 ID
MODEL_ID = "gemini-2.0-flash-001"

def get_access_token():
    """gcloud CLI로 Vertex AI용 Access Token을 가져옴."""
    try:
        token = subprocess.check_output(
            ["gcloud", "auth", "print-access-token"],
            text=True
        ).strip()
        print("✅ Access Token을 성공적으로 가져옴.")
        return token
    except (subprocess.CalledProcessError, FileNotFoundError) as e:
        print(f"❌ Access Token 획득 실패: {e}")
        return None

def ask_gemini_via_domain(prompt: str):
    """도메인(프라이빗 레졸버)으로 Vertex AI Gemini API 호출"""
    token = get_access_token()
    if not token:
        return
    
    # PSC를 통한 프라이빗 엔드포인트로 도메인 접속
    api_endpoint = (
        f"https://{VERTEX_API_DOMAIN}/v1/projects/{PROJECT_ID}"
        f"/locations/{LOCATION}/publishers/google/models/{MODEL_ID}:generateContent"
    )
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }
    request_body = {
        "contents": [
            {
                "role": "user",
                "parts": [
                    {"text": prompt}
                ]
            }
        ]
    }

    print(f"\n프라이빗 도메인({VERTEX_API_DOMAIN})으로 요청 중...")

    warnings.filterwarnings('ignore', category=InsecureRequestWarning)
    try:
        response = requests.post(
            api_endpoint, headers=headers, json=request_body,
            verify=False, timeout=20
        )
        response.raise_for_status()
        response_json = response.json()
        text_content = response_json["candidates"][0]["content"]["parts"][0]["text"]
        print("\n--- Gemini 응답 ---\n" + text_content + "\n--------------------")
        print("\n✅ 도메인 경유 요청 성공!")
    except requests.exceptions.HTTPError as e:
        print(f"\n❌ API 요청 시 HTTP 오류: {e.response.status_code}")
        print("   오류 내용:")
        print(json.dumps(e.response.json(), indent=2))
    except Exception as e:
        print(f"\n❌ 예상치 못한 오류: {e}")
    finally:
        warnings.resetwarnings()

if __name__ == "__main__":
    prompt_to_send = "자기소개 부탁해"
    ask_gemini_via_domain(prompt_to_send)

 

저작자표시 비영리 변경금지 (새창열림)

'Cloud > GCP' 카테고리의 다른 글

Slack 기반 GCE Managed Instance Group 오토스케일링 자동화 구축 가이드 (Pub/Sub 연동, 상태 알림 포함)  (4) 2025.08.08
GCP 프로젝트 로그 분석 리포트 생성 자동화하기 – n8n, AI Agent, Gemini, MCP Server (GCP)  (4) 2025.07.31
매일 GCP 프로젝트 로그 리포트 자동화하기 – n8n, AI Agent, Gemini, MCP Server (GCP)  (2) 2025.07.20
GCP IAP를 통한 어플리케이션(Web) 접근관리방안 feat. IAP + Cloud Run + Cloud Storage  (1) 2025.07.17
Gemini CLI와 GCP MCP Server 를 통한 GCP BIgquery 조회  (0) 2025.07.10
Gemini CLI와 GCP MCP Server 를 통한 Google Cloud 관리방안  (4) 2025.07.02
PSC 연결을 통한 Cloud Run to Vertex AI Gemini API 호출방안  (1) 2025.06.28

'Cloud/GCP'의 다른글

  • 현재글GCP VPN과 PSC를 통한 Vertex AI (Gemini-API) Private 연결 가이드

관련글

댓글

티스토리툴바