GCE에서 Cloud Run(Function) 호출 테스트

 

GCP GCE에서 Serverless 서비스인 Cloud Run (또는 Cloud Functions)을 호출하는 방안에 대해 기술합니다.
Cloud Run은 자동 스케일링, 관리 편의성 등의 장점이 있어, 유연하고 확장성 있는 서비스를 제공하는 목적에 적합한 선택입니다.

이번 테스트에서는 GCE 인스턴스에서 Cloud Run(Function)을 호출하는 과정을 검증합니다.
1. GCE 인스턴스에서 GCP Metadata 서버를 통해 ID 토큰을 요청하여 인증 정보를 획득합니다.
2. 획득한 인증 토큰을 사용하여 Cloud Run(Function)을 호출합니다.
3. Function 내에서 GCE의 hostname을 수신 및 확인하고, 응답을 GCE로 반환합니다.
4. Cloud Logging을 통해 실행 여부 및 결과를 모니터링

 

 

전체 흐름도

 

1. Cloud Run (function) 생성

소스코드
함수 진입점 : show_hostname

import functions_framework
import flask

@functions_framework.http
def show_hostname(request: flask.Request) -> flask.Response:
    """
    HTTP Cloud Function that receives a hostname and displays it.
    Expects a JSON payload with a "hostname" key,
    or a query parameter "hostname".
    """
    hostname = None
    if request.method == "POST":
        try:
            data = request.get_json(silent=True)
            if data and "hostname" in data:
                hostname = data["hostname"]
        except Exception as e:
            print(f"Error parsing JSON: {e}") # 로깅

    if not hostname and "hostname" in request.args:
        hostname = request.args.get("hostname")

    if hostname:
        return flask.jsonify({"received_hostname": hostname})
    else:
        return flask.jsonify({"error": "Hostname not provided. Please send it as a JSON payload {'hostname': 'your-gce-hostname'} or as a query parameter ?hostname=your-gce-hostname"}), 400

 

2. GCE설정
Cloud Run (function) 호출하는 코드

import socket
import requests
import os

def get_hostname():
    """Retrieves the hostname of the current machine."""
    return socket.gethostname()

def send_hostname_to_cloud_function(function_url, hostname_to_send):
    """Sends the hostname to the specified Cloud Function URL."""
    headers = {"Content-Type": "application/json"}
    payload = {"hostname": hostname_to_send}

    # --- ID 토큰 가져오기 시작 ---
    # GCE 메타데이터 서버에서 ID 토큰을 가져옵니다.
    # audience는 호출하려는 Cloud Function의 URL이어야 합니다.
    metadata_server_token_url = f"http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/identity?audience={function_url}"
    token_response = requests.get(metadata_server_token_url, headers={"Metadata-Flavor": "Google"})

    try:
        token_response.raise_for_status() # 오류 발생 시 예외 처리
        id_token = token_response.text
        headers["Authorization"] = f"Bearer {id_token}"
        print("ID Token successfully fetched and added to headers.") # 토큰 발급 성공 로그
    except requests.exceptions.RequestException as e:
        print(f"Error fetching ID token: {e}")
        print(f"Token response content: {token_response.content}") # 토큰 응답 내용 확인
        return # 토큰 발급 실패 시 함수 종료
    # --- ID 토큰 가져오기 끝 ---

    try:
        response = requests.post(function_url, json=payload, headers=headers, timeout=10)
        response.raise_for_status()  # Raises an exception for HTTP errors
        print(f"Cloud Function Response: {response.json()}")
    except requests.exceptions.RequestException as e:
        print(f"Error calling Cloud Function: {e}")
        if hasattr(e, 'response') and e.response is not None: # 응답 객체가 있는지 확인
             print(f"Cloud Function error response: {e.response.text}") # 함수 에러 응답 확인

if __name__ == "__main__":
    # Cloud Function 배포 후 얻은 HTTPS 트리거 URL로 변경하세요.
    # 예시 URL: https://functi.asia-northeast3.run.app/
    cloud_function_trigger_url = "Cloud Function의 트리거 URL/" # 사용자 환경에 맞게 수정

    if "YOUR_CLOUD_FUNCTION_TRIGGER_URL" in cloud_function_trigger_url: # 기본값 체크 보강
        print("오류: Cloud Function의 트리거 URL을 스크립트에 정확히 설정해주세요.")
    else:
        current_hostname = get_hostname()
        print(f"현재 GCE VM의 호스트 이름: {current_hostname}")
        send_hostname_to_cloud_function(cloud_function_trigger_url, current_hostname)

 

3. 실행결과

Cloud Funtion 로그
호출 되었음을 알 수 있습니다.