مستوى المقال: متوسط
لو فاتورة AWS الشهر اللي فات قفزت من 1,800 دولار لـ 4,200 دولار وانت عرفت ده يوم 28 من الشهر، انت بتدفع ضريبة latency على الـ monitoring مش على الفاتورة نفسها. سكربت Python في 60 سطر مع Lambda و EventBridge بيكتشف القفزة في نفس يومها ويبعت تنبيه Slack قبل ما الفاتورة تكمل الشهر. تكلفة الحل الكلية أقل من 0.50 دولار شهرياً.
ليه فاتورة AWS بتفاجئك في آخر الشهر؟
المشكلة مش في حجم الإنفاق. المشكلة في الـ feedback loop المتأخر. لمّا تشوف الفاتورة آخر الشهر بس، أي قفزة بدأت يوم 5 بتكون اتراكمت 25 يوم قبل ما تلاحظ. الفرق بين كشف يومي وكشف شهري في خدمة GPU منسية ممكن يبقى 9,000 دولار في حادث واحد.
المثال اللي يقرّب الفكرة قبل التعريف العلمي
تخيل صاحب محل بقالة بيشوف الإيراد آخر الشهر بس. حد سرق 200 جنيه يومياً لمدة أسبوع، لكنه ما يلاحظش إلا لمّا يعمل الجرد آخر الشهر. الفلوس راحت من 7 أيام، ومافيش طريقة يسترجعها. الـ feedback loop المتأخر هو اللي خلّى السرقة تتراكم، مش حجم السرقة اليومي.
نفس الفكرة في AWS. لو نسيت EC2 instance شغّال في region تاني، أو فعّلت debug logging على CloudWatch، أو سيرفر CI بيعمل re-build كل دقيقة في loop، الفاتورة بتزيد في يومها لكن انت ما تشوفهاش غير في billing dashboard آخر الشهر. المشكلة هي الـ delay، مش الإنفاق.
تعريف Anomaly Detection على بيانات التكلفة
Anomaly Detection على سلسلة زمنية بمعنى علمي بسيط: هل القيمة في اليوم ده بعيدة بشكل غير طبيعي عن النمط المتوقع؟ فيه طريقتين شائعتين للحالة دي:
- Z-score: تحسب متوسط آخر N يوم وانحراف معياري، وتعتبر اليوم anomaly لو القيمة بعيدة بأكتر من 2.5 انحراف معياري عن المتوسط. سهل، شفاف، وكفاية في 90% من الحالات.
- Random Cut Forest (RCF): خوارزمية AWS بيستخدمها داخلياً في خدمة Cost Anomaly Detection الرسمية، مبنية على ورقة Guha et al. ICML 2016. أقوى مع seasonality لكن أصعب في الـ tuning.
للـ baseline detection اليومي، Z-score كفاية. RCF بيتفوّق لو الـ traffic عندك فيه أنماط أسبوعية واضحة (السبت أعلى من باقي الأيام مثلاً)، وحتى ساعتها AWS بيوفّر الخدمة دي مجاناً ومش محتاج تبنيها بنفسك.
السكربت التنفيذي - Lambda + Cost Explorer + Slack
السكربت بيستخدم Cost Explorer API (ce.get_cost_and_usage) ويبعت تنبيه على Slack webhook. شغّاله على Lambda بـ EventBridge schedule يومي 8 الصبح UTC. الافتراض: عندك حساب AWS واحد و Slack workspace.
import boto3
import json
import statistics
import urllib.request
from datetime import datetime, timedelta
import os
SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK"]
Z_THRESHOLD = 2.5
LOOKBACK_DAYS = 14
def lambda_handler(event, context):
ce = boto3.client("ce", region_name="us-east-1")
end = datetime.utcnow().date()
start = end - timedelta(days=LOOKBACK_DAYS)
response = ce.get_cost_and_usage(
TimePeriod={"Start": str(start), "End": str(end)},
Granularity="DAILY",
Metrics=["UnblendedCost"],
GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}],
)
daily_totals = {}
for day in response["ResultsByTime"]:
date = day["TimePeriod"]["Start"]
total = sum(
float(g["Metrics"]["UnblendedCost"]["Amount"])
for g in day["Groups"]
)
daily_totals[date] = total
sorted_days = sorted(daily_totals.items())
history = [v for _, v in sorted_days[:-1]]
today_value = sorted_days[-1][1]
today_date = sorted_days[-1][0]
if len(history) < 7:
return {"status": "not_enough_data"}
mean = statistics.mean(history)
stdev = statistics.stdev(history) or 1.0
z_score = (today_value - mean) / stdev
if z_score > Z_THRESHOLD:
msg = (
f":rotating_light: AWS Cost Anomaly\n"
f"Date: {today_date}\n"
f"Today: ${today_value:.2f}\n"
f"Avg(14d): ${mean:.2f}\n"
f"Z-score: {z_score:.2f}"
)
req = urllib.request.Request(
SLACK_WEBHOOK,
data=json.dumps({"text": msg}).encode(),
headers={"Content-Type": "application/json"},
)
urllib.request.urlopen(req, timeout=5)
return {"status": "ok", "z_score": round(z_score, 2)}
