OS: macOS 14.3
Google Cloud SDK: 539.0.0
Gemini CLI: 0.17.1
今天要來寫 Cloud Run Job 執行小記
在使用 Cloud Run 服務的時候, 會有兩種類型的執行方式, 分別是
Cloud Run Service
Cloud Run Job
這邊使用 NotebookLM 參考官方文件, 先將 Service 與 Job 大概區分出來, 如以下圖示
以自己Infra的角度來說, Cloud Run 如果是自己的應用面, 可能有兩個切入點
Cloud Run Service: 將要執行的工作建立成服務, 然後藉由不同的傳入值與呼叫不同的 API 達成目的.
Cloud Run Job: 將要執行的工作封裝為容器, 手動執行或是編為排程.
依照我個人來說, Cloud Run Job 比較適合我的維運場景, 故今天來寫這篇 Cloud Run Job 執行小記, 執行想法如下
目的: 使用 Google Cloud Recommender API 抓取虛擬機 (VM) 的大小調整建議
使用 Gemini CLI 產生 Python 檔案
使用 Gemini CLI - 將剛剛建立的 Python 檔案封裝至容器
使用 Gemini CLI - 建立 Shell script 執行 Cloud Run Job 部署
使用 Gemini CLI - 建立說明檔
首先建立 python 檔案如下
==== config.json 檔案 ====
{
"projects": [
"your-project-id-1"
],
"locations": [
"asia-east1-c",
"us-central1",
"europe-west1-b"
],
"gcs_bucket_name_for_table": "your-bucket-for-text-reports",
"gcs_bucket_name_for_json": "your-bucket-for-json-data",
"gcs_bucket_project": "your-gcs-bucket-project-id",
"gcs_bucket_location": "asia-east1"
}
==== pyproject.toml 檔案 ====
[project]
name = "temp"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"google-cloud-recommender>=2.18.2",
"google-cloud-storage>=3.4.1",
]
==== recommend.py 檔案 ====
import json
from datetime import datetime
from google.cloud import recommender_v1, storage
# --- 設定區塊 --- #
# 載入設定檔 config.json
def load_config():
"""從 config.json 載入設定參數"""
try:
with open("config.json", "r") as f:
config = json.load(f)
# 回傳從 config.json 讀取到的各項設定
return (
config["projects"], # GCP 專案 ID 列表
config["locations"], # GCP 地區列表
config.get("gcs_bucket_name_for_table"), # 用於儲存文字報告的 GCS 儲存桶名稱
config.get("gcs_bucket_name_for_json"), # 用於儲存 JSON 報告的 GCS 儲存桶名稱
config.get("gcs_bucket_project"), # 建立 GCS 儲存桶的專案 ID
config.get("gcs_bucket_location"), # 建立 GCS 儲存桶的區域
)
except FileNotFoundError:
print("錯誤：找不到 config.json 檔案。請確認檔案是否存在。")
exit(1)
except KeyError:
print("錯誤：config.json 檔案格式不正確，缺少必要的鍵。請檢查檔案內容。")
exit(1)
# 載入設定並賦值給全域變數
PROJECTS, LOCATIONS, GCS_BUCKET_TABLE, GCS_BUCKET_JSON, GCS_BUCKET_PROJECT, GCS_BUCKET_LOCATION = load_config()
RECOMMENDER_ID = "google.compute.instance.MachineTypeRecommender" # Recommender ID，用於取得 VM 大小調整建議
# --- 設定區塊結束 --- #
def generate_table_output(recommendations):
"""產生人類可讀的表格格式報告 (字串)
Args:
recommendations (list): 包含推薦資料的列表。
Returns:
str: 格式化的表格報告字串。
"""
if not recommendations:
return "找不到任何推薦。"
lines = []
# 動態計算欄寬，讓 summary 可以完整顯示
project_w = max([len(r['project']) for r in recommendations] + [len('Project')])
location_w = max([len(r['location']) for r in recommendations] + [len('Location')])
vm_name_w = max([len(r['vm_name']) for r in recommendations] + [len('VM Name')])
savings_w = 15 # 固定寬度，用於顯示節省金額
summary_w = max([len(r['summary']) for r in recommendations] + [len('Summary')])
# 表格標頭
header = f"| {'Project':<{project_w}} | {'Location':<{location_w}} | {'VM Name':<{vm_name_w}} | {'Savings (USD)':>{savings_w}} | {'Summary':<{summary_w}} |"
separator = '-' * len(header)
lines.append(separator)
lines.append(header)
lines.append(separator)
# 逐行添加推薦資料
for rec in recommendations:
row = f"| {rec['project']:<{project_w}} | {rec['location']:<{location_w}} | {rec['vm_name']:<{vm_name_w}} | {rec['monthly_savings_usd']:>{savings_w}.2f} | {rec['summary']:<{summary_w}} |"
lines.append(row)
lines.append(separator)
return "\n".join(lines)
def upload_to_gcs(storage_client, bucket_name, filename, content, content_type):
"""上傳字串內容到指定的 GCS 儲存桶
Args:
storage_client (google.cloud.storage.Client): GCS 客戶端實例。
bucket_name (str): 目標儲存桶的名稱。
filename (str): 上傳到 GCS 後的檔案名稱。
content (str): 要上傳的字串內容。
content_type (str): 內容類型 (例如 'text/plain', 'application/json')。
"""
try:
bucket = storage_client.bucket(bucket_name)
blob = bucket.blob(filename)
blob.upload_from_string(content, content_type=content_type)
print(f"成功！檔案已上傳到 gs://{bucket_name}/{filename}")
except Exception as e:
print(f"上傳到 gs://{bucket_name} 失敗: {e}")
print("請確認儲存桶名稱是否正確，以及您是否有寫入權限。")
def create_bucket_if_not_exists(storage_client, bucket_name, location, project_id):
"""檢查儲存桶是否存在，若不存在則嘗試建立。
Args:
storage_client (google.cloud.storage.Client): GCS 客戶端實例。
bucket_name (str): 要檢查或建立的儲存桶名稱。
location (str): 儲存桶的地理位置 (region)。
project_id (str): 儲存桶所屬的專案 ID。
Returns:
google.cloud.storage.Bucket: 儲存桶實例，如果建立失敗則為 None。
"""
try:
bucket = storage_client.lookup_bucket(bucket_name)
if bucket is None:
print(f"儲存桶 gs://{bucket_name} 不存在，正在嘗試於 {location} 建立...")
# 嘗試在指定的專案和位置建立儲存桶
new_bucket = storage_client.create_bucket(bucket_name, project=project_id, location=location)
print(f"成功建立儲存桶 gs://{new_bucket.name}。")
return new_bucket
return bucket
except Exception as e:
print(f"建立或檢查儲存桶 gs://{bucket_name} 時發生錯誤: {e}")
print("請確認您有足夠的權限 (例如 roles/storage.admin) 來建立儲存桶。")
return None
def fetch_and_process_recommendations():
"""抓取推薦、處理並上傳到 GCS。
此函數會執行以下步驟：
1. 初始化 Recommender 客戶端。
2. 遍歷 config.json 中定義的每個專案和地區，獲取 VM 大小調整建議。
3. 從推薦結果中提取 VM 名稱、摘要、預估每月節省金額等資訊。
4. 將所有推薦資訊彙整成列表。
5. 如果有推薦，則生成表格格式和 JSON 格式的報告。
6. 將表格報告印在終端機上供預覽。
7. 使用指定的專案和位置，確保 GCS 儲存桶存在（如果不存在則建立）。
8. 將兩種格式的報告上傳到 GCS。
"""
recommender_client = recommender_v1.RecommenderClient() # 初始化 Recommender 客戶端
all_recommendations = [] # 用於存放所有專案和地區的推薦資訊
print("開始掃描推薦...")
for project in PROJECTS:
for location in LOCATIONS:
# 構建 Recommender API 的父路徑
parent = f"projects/{project}/locations/{location}/recommenders/{RECOMMENDER_ID}"
print(f"正在取得 {parent} 的推薦...")
try:
# 列出指定路徑下的所有推薦
for rec in recommender_client.list_recommendations(parent=parent):
instance_name = "N/A"
try:
# 從推薦內容中提取 VM 實例名稱
for op_group in rec.content.operation_groups:
for op in op_group.operations:
if "/instances/" in op.resource:
instance_name = op.resource.split("/instances/")[-1]
break
if instance_name != "N/A": break
except (AttributeError, IndexError): pass
savings_amount = 0.0
try:
# 從推薦中提取預估的每月節省金額
cost = rec.primary_impact.cost_projection.cost
if cost.units or cost.nanos: savings_amount = -1 * (cost.units + cost.nanos / 1e9)
except (AttributeError, IndexError): pass
# 整理推薦數據
recommendation_data = {
"vm_name": instance_name,
"summary": rec.description,
"monthly_savings_usd": round(savings_amount, 2),
"project": project,
"location": location,
"recommendation_name": rec.name,
}
all_recommendations.append(recommendation_data)
except Exception as e:
print(f"無法從 {parent} 取得推薦: {e}")
if not all_recommendations:
print("找不到任何推薦，無需上傳。")
return
print(f"\n共找到 {len(all_recommendations)} 則推薦。")
# 1. 產生兩種格式的報告
table_report_string = generate_table_output(all_recommendations)
json_report_string = json.dumps(all_recommendations, indent=2, ensure_ascii=False)
# 2. 將表格格式印在終端機預覽
print("--- 表格報告 (預覽) ---")
print(table_report_string)
print("--- 結束預覽 ---")
# 3. 上傳兩種格式到 GCS
# 初始化 GCS 客戶端，並指定專案 ID
storage_client = storage.Client(project=GCS_BUCKET_PROJECT)
current_date = datetime.now().strftime('%Y-%m-%d') # 獲取當前日期作為檔案名稱的一部分
# 設定儲存桶的位置，優先使用 config.json 中的 GCS_BUCKET_LOCATION，否則使用第一個推薦地區，最後預設為 'US'
bucket_location = GCS_BUCKET_LOCATION if GCS_BUCKET_LOCATION else (LOCATIONS[0] if LOCATIONS else "US")
# 如果設定了文字報告的 GCS 儲存桶名稱
if GCS_BUCKET_TABLE:
# 檢查或建立儲存桶
if create_bucket_if_not_exists(storage_client, GCS_BUCKET_TABLE, bucket_location, GCS_BUCKET_PROJECT):
table_filename = f"text-reports/recommendations_{current_date}.txt"
print(f"\n正在上傳表格報告至 gs://{GCS_BUCKET_TABLE}/{table_filename}...")
upload_to_gcs(storage_client, GCS_BUCKET_TABLE, table_filename, table_report_string, "text/plain; charset=utf-8")
# 如果設定了 JSON 報告的 GCS 儲存桶名稱
if GCS_BUCKET_JSON:
# 檢查或建立儲存桶
if create_bucket_if_not_exists(storage_client, GCS_BUCKET_JSON, bucket_location, GCS_BUCKET_PROJECT):
json_filename = f"json-data/recommendations_{current_date}.json"
print(f"\n正在上傳 JSON 報告至 gs://{GCS_BUCKET_JSON}/{json_filename}...")
upload_to_gcs(storage_client, GCS_BUCKET_JSON, json_filename, json_report_string, "application/json")
# 腳本主入口點
if __name__ == "__main__":
fetch_and_process_recommendations()
建立完成後進入到下一個階段, 請 gemini cli 封裝成容器
==== Dockerfile 檔案 ====
# Use an official Python runtime as a parent image
FROM python:3.12-slim
# Set the working directory in the container
WORKDIR /app
# Copy pyproject.toml and install dependencies
# This ensures that dependencies are installed before copying the rest of the application
# which can help with Docker layer caching.
COPY pyproject.toml pyproject.toml
# Install build dependencies and then project dependencies
RUN pip install --no-cache-dir ".[dev]" && \
pip install --no-cache-dir .
# Copy the rest of the application code
COPY . .
# Set the command to run the application
CMD ["python", "recommend.py"]
接下來請 Gemini CLI 建立一個 deploy.sh 來協助部署
==== deploy.sh 檔案 ====
#!/bin/bash
# -----------------------------------------------------------------------------
# GCP Cloud Run Job 部署腳本 - GCP Recommender
# -----------------------------------------------------------------------------
# 這個腳本用於自動化建置 Docker 映像檔、將其推送到 Google Artifact Registry，
# 並部署為 Google Cloud Run Job。此 Job 的主要功能是根據 config.json 的設定，
# 抓取 GCP Recommender 的建議，並將報告上傳到 GCS。
#
# 此腳本將使用 GCP 專案的「預設 Compute Engine 服務帳戶」來執行 Job。
#
# 使用前請務必修改「使用者可配置變數」區塊中的值。
# -----------------------------------------------------------------------------
# --- 使用者可配置變數 ---
# -----------------------------------------------------------------------------
# 請根據您的 GCP 環境和需求修改以下變數。
# -----------------------------------------------------------------------------
# GCP_PROJECT_ID:
# 這是您的 GCP 專案 ID。此專案將用於：
# 1. 建置 Docker 映像檔 (透過 Cloud Build)。
# 2. 存放 Docker 映像檔 (在 Artifact Registry 中)。
# 3. 部署 Cloud Run Job。
# 請將 "your-gcp-project-id" 替換為您實際的專案 ID。
GCP_PROJECT_ID="your-gcp-project-id"
# GCP_REGION:
# Cloud Run Job 將被部署到的 GCP 區域。
# 範例: "asia-east1", "us-central1"
GCP_REGION="asia-east1"
# REPO_NAME:
# 用於存放 Docker 映像檔的 Artifact Registry 倉庫名稱。
# 如果倉庫不存在，腳本會嘗試建立它。
REPO_NAME="cloud-run-jobs"
# IMAGE_NAME:
# 您要為 Docker 映像檔指定的名稱。
IMAGE_NAME="gcp-recommender"
# JOB_NAME:
# 您要為 Cloud Run Job 指定的名稱。
JOB_NAME="gcp-recommender-job"
# -----------------------------------------------------------------------------
# --- 內部變數與邏輯 ---
# -----------------------------------------------------------------------------
IMAGE_TAG="${GCP_REGION}-docker.pkg.dev/${GCP_PROJECT_ID}/${REPO_NAME}/${IMAGE_NAME}:latest"
# -----------------------------------------------------------------------------
# --- 步驟 1: 檢查並啟用必要的 GCP APIs ---
# -----------------------------------------------------------------------------
echo ""
echo "\n--- 正在檢查並啟用必要的 GCP APIs... ---"
echo ""
# 需要的 API 列表
REQUIRED_APIS=(
"cloudbuild.googleapis.com"
"run.googleapis.com"
"recommender.googleapis.com"
"storage.googleapis.com"
"artifactregistry.googleapis.com"
)
# 獲取已啟用的 API 列表
ENABLED_APIS=$(gcloud services list --enabled --project="${GCP_PROJECT_ID}" --format="value(config.name)")
for API in "${REQUIRED_APIS[@]}"; do
if [[ ! " ${ENABLED_APIS[@]} " =~ " ${API} " ]]; then
echo "正在啟用 API: ${API}..."
gcloud services enable "${API}" --project="${GCP_PROJECT_ID}"
if [ $? -ne 0 ]; then
echo "\n❌ 啟用 API '${API}' 失敗。請檢查權限或手動啟用後再試。"
exit 1
fi
echo "✅ API '${API}' 已啟用。"
else
echo "✅ API '${API}' 已啟用。"
fi
done
# -----------------------------------------------------------------------------
# --- 步驟 2: 檢查並建立 Artifact Registry 倉庫 ---
# -----------------------------------------------------------------------------
echo ""
echo "\n--- 正在檢查並建立 Artifact Registry 倉庫... ---"
echo ""
# 檢查倉庫是否存在
if ! gcloud artifacts repositories describe "${REPO_NAME}" --location="${GCP_REGION}" --project="${GCP_PROJECT_ID}" &> /dev/null; then
echo "倉庫 '${REPO_NAME}' 不存在，正在於區域 '${GCP_REGION}' 中建立..."
gcloud artifacts repositories create "${REPO_NAME}" \
--repository-format=docker \
--location="${GCP_REGION}" \
--project="${GCP_PROJECT_ID}" \
--description="Docker repository for Cloud Run Jobs"
if [ $? -ne 0 ]; then
echo "\n❌ 建立 Artifact Registry 倉庫 '${REPO_NAME}' 失敗。請檢查權限或手動建立後再試。"
exit 1
fi
echo "✅ 倉庫 '${REPO_NAME}' 已成功建立。"
else
echo "✅ 倉庫 '${REPO_NAME}' 已存在。"
fi
# -----------------------------------------------------------------------------
# --- 步驟 3: 執行前確認 ---
# -----------------------------------------------------------------------------
echo ""
echo "將要執行以下操作:"
echo "1. 在專案 '${GCP_PROJECT_ID}' 中建置 Docker 映像檔並推送到 Artifact Registry"
echo " - 映像檔路徑: ${IMAGE_TAG}"
echo ""
echo "2. 部署/更新 Cloud Run Job '${JOB_NAME}'"
echo " - 部署區域: ${GCP_REGION}"
echo " - 服務帳戶: 將使用專案預設的 Compute Engine 服務帳戶"
echo " (請確保此帳戶有足夠的 IAM 權限，詳見 CLOUD_RUN_JOB_DEPLOYMENT.md)"
# 提示使用者確認是否繼續
echo ""
read -p "確定要繼續嗎？ (y/n) " -n 1 -r
echo
if [[ ! $REPLY =~ ^[Yy]$ ]]
then
[[ "$0" = "$BASH_SOURCE" ]] && exit 1 || return 1
fi
# -----------------------------------------------------------------------------
# --- 步驟 4: 使用 Cloud Build 建置並推送映像檔 ---
# -----------------------------------------------------------------------------
echo ""
echo "\n--- 正在建置 Docker 映像檔並推送到 Artifact Registry... ---"
echo ""
gcloud builds submit --tag "${IMAGE_TAG}" . --project="${GCP_PROJECT_ID}"
if [ $? -ne 0 ]; then
echo "\n❌ 映像檔建置或推送失敗。請檢查錯誤訊息。"
exit 1
fi
echo "✅ 映像檔建置並推送成功: ${IMAGE_TAG}"
echo ""
# -----------------------------------------------------------------------------
# --- 步驟 5: 部署 Cloud Run Job ---
# -----------------------------------------------------------------------------
echo ""
echo "\n--- 正在部署 Cloud Run Job '${JOB_NAME}'... ---"
echo ""
gcloud run jobs deploy "${JOB_NAME}" \
--image "${IMAGE_TAG}" \
--region "${GCP_REGION}" \
--cpu 1 \
--memory "512Mi" \
--max-retries 0 \
--task-timeout "3600s" \
--project "${GCP_PROJECT_ID}"
if [ $? -eq 0 ]; then
echo ""
echo "\n✅ Cloud Run Job '${JOB_NAME}' 部署成功！"
echo ""
echo "您可以執行以下指令來手動觸發 Job 執行:"
echo "gcloud run jobs execute ${JOB_NAME} --region ${GCP_REGION} --project ${GCP_PROJECT_ID}"
echo ""
echo "\n若要查看 Job 的執行結果，請使用以下指令:"
echo "gcloud logging read 'resource.type=\"cloud_run_job\" AND resource.labels.job_name=\"${JOB_NAME}\"' --project ${GCP_PROJECT_ID} --format 'value(textPayload)' --limit=50"
echo ""
else
echo "\n❌ Cloud Run Job '${JOB_NAME}' 部署失敗。請檢查錯誤訊息。"
fi
# -----------------------------------------------------------------------------
最後就是執行 deploy.sh 來進行部署了 :)
好像又前進一步
~ enjoy it
References
