GCE 소스
import socket
import requests
def get_gce_metadata(metadata_path):
"""GCE 메타데이터 서버에서 정보를 가져오는 함수"""
metadata_server_url = f"http://metadata.google.internal/computeMetadata/v1/{metadata_path}"
headers = {"Metadata-Flavor": "Google"}
try:
response = requests.get(metadata_server_url, headers=headers, timeout=5)
response.raise_for_status()
return response.text
except requests.exceptions.RequestException as e:
print(f"Error fetching metadata from {metadata_path}: {e}")
return None
def send_info_to_cloud_function(function_url, payload):
"""주어진 정보를 Cloud Function으로 전송하는 함수"""
headers = {"Content-Type": "application/json"}
# --- ID 토큰 가져오기 ---
audience = function_url
id_token = get_gce_metadata(f"instance/service-accounts/default/identity?audience={audience}")
if not id_token:
print("Failed to get ID token. Aborting.")
return
headers["Authorization"] = f"Bearer {id_token}"
print("✅ ID Token successfully fetched and added to headers.")
# --- Cloud Function 호출 ---
try:
response = requests.post(function_url, json=payload, headers=headers, timeout=10)
response.raise_for_status()
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__":
# 1. Cloud Function URL을 설정합니다. (실제 URL로 변경필요)
cloud_function_trigger_url = "YOUR_CLOUD_FUNCTION_URL"
if "YOUR_CLOUD_FUNCTION_URL" in cloud_function_trigger_url:
print("❌ 오류: Cloud Function의 트리거 URL을 스크립트에 정확히 설정필요")
else:
# 2. GCE VM의 정보를 수집합니다.
hostname = socket.gethostname()
internal_ip = get_gce_metadata("instance/network-interfaces/0/ip")
external_ip = get_gce_metadata("instance/network-interfaces/0/access-configs/0/external-ip")
print("--- GCE VM Info ---")
print(f"Hostname: {hostname}")
print(f"Internal IP: {internal_ip}")
print(f"External IP: {external_ip}")
print("--------------------")
# 3. Cloud Function에 보낼 데이터 '꾸러미(payload)'를 만듭니다.
payload_to_send = {
"gce_hostname": hostname,
"gce_internal_ip": internal_ip,
"gce_external_ip": external_ip
}
# 4. 정보를 담아 Cloud Function을 호출합니다.
send_info_to_cloud_function(cloud_function_trigger_url, payload_to_send)
Cloud Run(Function)소스
import functions_framework
import flask
@functions_framework.http
def log_caller_info(request: flask.Request) -> flask.Response:
"""HTTP Cloud Function that logs caller's info and returns it."""
# --- 1. 요청 헤더에서 호출자 정보 확인 ---
# request.headers는 '편지 봉투'에 적힌 발신자 정보와 같아요.
user_agent = request.headers.get('User-Agent')
# 'X-Forwarded-For'는 로드밸런서를 거쳐올 때 실제 클라이언트 IP를 담고 있어요.
# GCE에서 호출했으므로, GCE의 외부 IP가 찍힐 거예요.
source_ip = request.headers.get('X-Forwarded-For')
print("--- [Request Info] ---")
print(f"Source IP (from X-Forwarded-For): {source_ip}")
print(f"User-Agent: {user_agent}")
# --- 2. GCE가 보낸 데이터(payload) 확인 ---
caller_data = {}
if request.method == "POST":
data = request.get_json(silent=True)
if data:
# GCE가 보낸 정보들을 '편지 내용'에서 꺼내옵니다.
caller_data['hostname'] = data.get('gce_hostname')
caller_data['internal_ip'] = data.get('gce_internal_ip')
caller_data['external_ip'] = data.get('gce_external_ip')
print("--- [Caller-Provided Info] ---")
print(f"GCE Hostname: {caller_data['hostname']}")
print(f"GCE Internal IP: {caller_data['internal_ip']}")
print(f"GCE External IP: {caller_data['external_ip']}")
# --- 3. 응답 보내기 ---
if caller_data.get('hostname'):
return flask.jsonify({
"status": "success",
"message": "Caller information received and logged.",
"received_data": caller_data
})
else:
return flask.jsonify({"error": "Required information not provided in payload."}), 400
