95 lines
2.7 KiB
Python
95 lines
2.7 KiB
Python
import base64
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import urllib.request
|
||
|
||
import functions_framework
|
||
from google.cloud import secretmanager
|
||
|
||
|
||
def get_webhook_url(project_id: str) -> str | None:
|
||
try:
|
||
client = secretmanager.SecretManagerServiceClient()
|
||
name = f"projects/{project_id}/secrets/webhook-url/versions/latest"
|
||
response = client.access_secret_version(request={"name": name})
|
||
return response.payload.data.decode("UTF-8")
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def send_webhook(url: str, message: str) -> None:
|
||
payload = json.dumps({"text": message}).encode()
|
||
req = urllib.request.Request(
|
||
url,
|
||
data=payload,
|
||
headers={"Content-Type": "application/json"},
|
||
method="POST",
|
||
)
|
||
with urllib.request.urlopen(req, timeout=10):
|
||
pass
|
||
|
||
|
||
@functions_framework.cloud_event
|
||
def billing_alert(cloud_event):
|
||
project_id = os.environ.get("PROJECT_ID", "")
|
||
service = os.environ.get("CLOUD_RUN_SERVICE", "")
|
||
region = os.environ.get("REGION", "")
|
||
|
||
# Dekod Pub/Sub-melding
|
||
data = base64.b64decode(cloud_event.data["message"]["data"]).decode()
|
||
alert = json.loads(data)
|
||
|
||
budget_amount = alert.get("budgetAmount", "?")
|
||
cost_amount = alert.get("costAmount", "?")
|
||
alert_threshold = alert.get("alertThresholdExceeded", 0)
|
||
|
||
print(f"Billing alert: {cost_amount} / {budget_amount} ({alert_threshold*100:.0f}%)")
|
||
|
||
webhook_url = get_webhook_url(project_id)
|
||
base_msg = (
|
||
f"🚨 *OSVauco Cost Alert*
|
||
"
|
||
f"Prosjekt: `{project_id}`
|
||
"
|
||
f"Forbruk: `{cost_amount}` av budsjett `{budget_amount}`
|
||
"
|
||
f"Terskel: `{alert_threshold*100:.0f}%`"
|
||
)
|
||
|
||
# Kjør auto-teardown kun ved 100% terskel
|
||
if alert_threshold >= 1.0:
|
||
print("100% terskel nådd — starter auto-teardown")
|
||
try:
|
||
subprocess.run(
|
||
[
|
||
"gcloud", "run", "services", "delete", service,
|
||
f"--region={region}",
|
||
f"--project={project_id}",
|
||
"--quiet",
|
||
],
|
||
check=True,
|
||
capture_output=True,
|
||
)
|
||
teardown_msg = base_msg + "
|
||
|
||
ℹ️ Auto-teardown utført: Cloud Run stoppet."
|
||
except subprocess.CalledProcessError as e:
|
||
teardown_msg = base_msg + f"
|
||
|
||
⚠️ Auto-teardown FEILET: {e.stderr.decode()}"
|
||
msg = teardown_msg
|
||
else:
|
||
msg = base_msg + "
|
||
|
||
⚠️ Ingen tiltak nå — men pass på forbruket."
|
||
|
||
if webhook_url:
|
||
try:
|
||
send_webhook(webhook_url, msg)
|
||
print("Webhook-alert sendt")
|
||
except Exception as ex:
|
||
print(f"Webhook feilet: {ex}")
|
||
else:
|
||
print("Ingen webhook-url i Secret Manager — hopper over varsling")
|