PSC 연결을 통한 Cloud Run to AI Studio Gemini API 호출방안
배경 및 목표
Cloud Run은 서버리스 컨테이너를 실행하기 위한 강력한 플랫폼이지만, 기본적으로 서비스 URL은 인터넷에 공개됩니다.
보안이 중요한 백엔드 API나 내부 시스템 간의 통신을 위해서는 서비스를 외부 인터넷으로부터 격리하고, 승인된 내부 네트워크에서만 안전하게 호출할 수 있는 아키텍처가 필수적입니다.본 문서는 Private Service Connect(PSC)와 내부 애플리케이션 부하 분산기(Internal ALB)를 조합하여, 공개 인터넷에 노출되지 않는 비공개 Cloud Run 서비스를 구축하는 과정을 다룹니다.
사용하는 Gemini API는 빠르게 최신기술을 적용할 수 있는 Public AI인 Google AI Studio 의 Gemini API를 이용합니다.
다음편에서는 동일한 SDK로 개발 가능하며, 기업을 위한 Vertex Gemini API를 이용하는 방안을 업로드 예정입니다.
AI Key발급하는 방안만 다를뿐 기능은 동일합니다.또한, 이 비공개 Cloud Run 서비스가 Gemini API와 연동하여 외부 서비스를 호출하는 End-to-End 흐름을 완성하는 것을 목표로 합니다.
- 목표 아키텍처
내부 네트워크 연결 방식으로는 VPN/VPC Peering과 PSC가 있습니다. 이 중 IP 충돌의 우려가 없고, 네트워크 단위가 아닌 서비스 단위로 명확하게 연결을 관리할 수 있어 보안 및 운영 편의성이 뛰어난 Private Service Connect(PSC)를 사용하는 아키텍처로 구성합니다
1. VPN or VPC Peering 으로 연결 하여 접속
2. Client VPC에 PSC Endpoint 생성하여 접속
- Service Project: 실제 Cloud Run 서비스와 Gemini API 연동 로직이 존재합니다. 내부 ALB와 PSC 서비스 게시
(Service Attachment)가 이곳에 구성됩니다. - Client Project: Cloud Run 서비스를 호출할 클라이언트(테스트 VM)가 존재합니다. PSC 엔드포인트가 이곳에 생성되며, VM은 이 엔드포인트의 내부 IP를 통해 Service Project의 Cloud Run 서비스를 호출합니다
본문서에서는 2안으로 진행합니다.
Clinet 용 VM → Cloud Run → Internal Load Balancer → PSC
구성절차
- Phase 1: Cloud Run 서비스 배포
- Google AI studio에 가서 Gemini API 키를 발급 받습니다.
https://aistudio.google.com/apikey
로그인 - Google 계정
이메일 또는 휴대전화
accounts.google.com
Gemini API를 호출하는 비즈니스 로직을 담은 Flask 애플리케이션을 준비하고, 이를 비공개 Cloud Run 서비스로 배포합니다.
main.py (소스코드)
import os
from flask import Flask, request, jsonify
import google.generativeai as genai
app = Flask(__name__)
# Cloud Run 배포 시 환경 변수로 API 키를 전달받음
try:
api_key = os.environ.get("API_KEY")
if not api_key:
raise ValueError("API_KEY environment variable not set.")
genai.configure(api_key=api_key)
# 모델 초기화는 한 번만 수행
model = genai.GenerativeModel('gemini-1.5-flash')
print("Gemini API client initialized successfully.")
except Exception as e:
print(f"FATAL: Failed to initialize Gemini API client: {e}")
raise e
@app.route("/", methods=['POST'])
def chat():
if not request.is_json or 'prompt' not in request.json:
return jsonify({"error": "Invalid request. Must be a POST request with a JSON body containing a 'prompt' key."}), 400
prompt = request.json['prompt']
try:
chat_session = model.start_chat()
response = chat_session.send_message(prompt)
print(f"Successfully received response from Gemini API for prompt: '{prompt[:30]}...'")
return jsonify({"response": response.text})
except Exception as e:
print(f"Error during Gemini API call: {e}")
return jsonify({"error": str(e)}), 500requirements.txt
Flask
google-generativeai
gunicornGUI로 생성 후 “새버전 수정 배포" → 컨테이너 → “변수 및 보안 비밀"에서
환경변수인 API KEY를 직접 추가 하거나, 아래의 gcloud CLI 명령어로 배포해도 됩니다.
# 아래 변수를 자신의 환경에 맞게 수정
PROJECT_ID="your-gcp-project-id"
REGION="asia-northeast3" # 예: 서울 리전
SERVICE_NAME="gemini-chat-api"
GEMINI_API_KEY="your_gemini_api_key"
# Cloud Run 배포 명령어
gcloud run deploy $SERVICE_NAME \
--project $PROJECT_ID \
--source . \
--region $REGION \
--allow-unauthenticated \
--set-env-vars="API_KEY=$GEMINI_API_KEY"
위와 같이 생성하면 환경 변수는 자동으로 삽입됩니다.
인그레스 및 인증 설정
배포 후, Cloud Run 서비스의 [네트워킹] 및 [보안] 탭에서 다음을 반드시 확인하고 설정합니다.
인그레스 제어:
내부를 선택하여 내부 트래픽만 허용하도록 설정합니다.
외부 애플리케이션 부하 분산기의 트래픽 허용 체크박스를 반드시 선택합니다. (이후 단계에서 만들 LB를 위함)
인증:
인증 필요를 선택하여 모든 요청에 IAM 기반의 유효한 ID 토큰이 포함되도록 강제합니다.
Cloud Run을 호출하기 위해서는 호출하는 Service Account 에 “Cloud Run 서비스 호출자" 권한이 있어야 됩니다.
Phase 2: 내부 애플리케이션 부하 분산기(ALB) 구성
내부에서 구성되는 ALB이므로 “애플리케이션 부하 분산기" → “내부" → “리전 워크로드"로 선택하여 구성 합니다.
또한 뒤에서 PSC를 사용하기 때문에 “내부 패스 스루 네트워크”(ILB), “리전별 내부 프록시 네트워크”(NLB), “리전별 내부 애플리케이션"만 선택이 가능합니다.
이중 ALB 종류는 “리전별 내부 애플리케이션”만 됩니다.
비공개 Cloud Run 서비스로 트래픽을 전달할 내부 ALB를 구성합니다.
부하 분산기 생성:
- 부하 분산기 유형: 애플리케이션 부하 분산기(HTTP/HTTPS)
- 공개 또는 내부: 내부
- 리전 간 또는 단일 리전 배포: 리전 워크로드에 적합 (리전별 내부 애플리케이션 부하 분산기)
백엔드 서비스 구성:
- 백엔드 유형: 서버리스 네트워크 엔드포인트 그룹(NEG)을 선택합니다.
- 새 백엔드 생성:
라우팅 및 프런트엔드 구성:
- 라우팅 규칙: 특별한 설정 없이 기본값으로 진행합니다.
- 프런트엔드 구성: 내부 ALB가 사용할 내부 IP 주소를 할당합니다. 이 IP는 이후 단계에서 직접 사용되지는 않습니다.
구성이 완료되면 Cloud Run 서비스가 백엔드로 연결된 내부 ALB가 생성됩니다.
각 부분의 상세한 내용은 다음과 같습니다.
부하 분산기 이름과 리전 그리고 ALB가 존재할 네트워크를 선택합니다.
backend service 구성
백엔드 유형에 “서버리스 네트워크 엔드포인트 그룹"을 선택합니다.
해당 부분을 선택하고 생성합니다.
그외 라우팅과 프런트엔드는 기본으로 설정하고 진행 합니다.
아래와 같이 Cloud Run이 연결된 Internal ALB 가 생성되었습니다.
Phase 3: Private Service Connect(PSC) 연결
PSC 생성 - 서비스 게시 (Service Attachment)
Internal ALB와 PSC를 서비스 게시로 연결 합니다.
세부정보로 선택것은
Internal ALB로 만들었기에 “부하 분산기" - “ 리전별 내부 애플리케이션 부하 분산기" 로 선택 합니다.
앞서 만든 ALB를 선택하고, PSC가 NAT 처리를 할 서브넷도 새로 생성 합니다.
본 문서에서는 원활한 테스트를 위해 연결 환경설정에는 “Automatically accept all connections” 로 선택하였으나 운영 환경에서는 환경에 맞도록 선택하시기를 바랍니다.
서비스를 게시하게 되면, ALB와 연결되면서 아래 화면과 같이 서비스 연결에 대한 주소가 발급됩니다.
생성 후 서비스 연결 주소를 복사합니다.
- PSC 생성 - 엔드포인트 연결 (End Point)
PSC IP를 통해 서비스를 연결하기 위해 설정합니다.
대상 세부정보에는 서비스 게시에서 설정된 “서비스 연결" : projects/testtttdf/regions/asia-northeast3/serviceAttachments/psc-to-lb 로 된 부분을 입력
이름과 PSC Endpoint 가 존재할 VPC, Subnet을 설정 후 PSC IP까지 할당합니다.
이제 192.168.1.150이란 PSC Endpoint IP가 생성되었습니다.
- Phase 4: 최종 테스트
모든 구성이 완료되었습니다. 이제 Client Project의 테스트 VM에서 Python 코드를 실행하여 최종적으로 API를 호출합니다.
PSC_ENDPOINT_URL, ORIGINAL_SERVICE_URL 변수를 확인하고 입력해줍니다.
만약 해당 VM에서 사용되는 SA에 “Cloud Run 서비스 호출자” 권한이 없다면 확인 후 권한부여를 합니다.
- 권한 부여
# Cloud Shell 또는 로컬 터미널에서 실행 SERVICE_NAME="gemini-public-test" REGION="asia-northeast3" VM_SERVICE_ACCOUNT="your-vm-service-account@...gserviceaccount.com" gcloud run services add-iam-policy-binding $SERVICE_NAME \ --project [서비스가 있는 프로젝트 ID] \ --region=$REGION \ --member="serviceAccount:$VM_SERVICE_ACCOUNT" \ --role="roles/run.invoker"테스트 클라이언트 코드
import os import json import requests import subprocess # 셸 명령어를 실행하기 위해 추가 # --- 설정값 --- # PSC를 통해 내부에서 접속할 IP 주소 PSC_ENDPOINT_URL = "http://192.168.1.150" ORIGINAL_SERVICE_URL = "https://gemini-public-test-654614075700.asia-northeast3.run.app" # Cloud Run 대표 URL로 해야됩니다. def get_identity_token_with_gcloud(audience_url): """ gcloud CLI를 직접 호출하여 ID 토큰을 생성합니다. """ try: command = [ "gcloud", "auth", "print-identity-token", "--audiences", audience_url ] result = subprocess.run(command, capture_output=True, text=True, check=True) return result.stdout.strip() except FileNotFoundError: print("❌ 오류: 'gcloud' 명령어를 찾을 수 없습니다. GCE VM에 Google Cloud CLI가 설치되어 있는지 확인하세요.") return None except subprocess.CalledProcessError as e: print(f"❌ gcloud 명령어로 토큰을 생성하는 데 실패했습니다:") print(f" Stderr: {e.stderr.strip()}") return None def start_chat_session(): """사용자와 Gemini 간의 대화 세션""" print("☁️ gcloud 명령어를 호출하여 인증 토큰을 생성하는 중...") identity_token = get_identity_token_with_gcloud(ORIGINAL_SERVICE_URL) if not identity_token: return print("✅ 인증 토큰을 성공적으로 생성했습니다.") headers = { "Content-Type": "application/json", "Host": ORIGINAL_SERVICE_URL.replace("https://", ""), "Authorization": f"Bearer {identity_token}" } print("✨ Chat with Gemini started! (Type 'exit' or 'quit' to end)") print("-" * 30) while True: try: prompt = input("You: ") if prompt.lower() in ["exit", "quit"]: print("👋 Chat session ended.") break payload = {"prompt": prompt} response = requests.post(PSC_ENDPOINT_URL, data=json.dumps(payload), headers=headers, timeout=60) response.raise_for_status() response_data = response.json() gemini_response = response_data.get("response") print(f"Gemini: {gemini_response}") except requests.exceptions.RequestException as e: print(f"❌ 요청 중 오류가 발생했습니다: {e}") if hasattr(e, 'response') and e.response is not None: print(f"Error details (HTTP {e.response.status_code}): {e.response.text}") break except KeyboardInterrupt: print("\n👋 Chat session ended.") break if __name__ == "__main__": start_chat_session()
최종 테스트 후 호출성공 화면
