86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
import os
|
|
import json
|
|
import datetime
|
|
from google.cloud import bigquery
|
|
from google.cloud import pubsub_v1
|
|
|
|
class AnomalyDetector:
|
|
def __init__(self):
|
|
self.project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
|
|
if not self.project_id:
|
|
raise ValueError("GOOGLE_CLOUD_PROJECT environment variable not set.")
|
|
|
|
self.pubsub_topic_name = os.environ.get("PUBSUB_TOPIC", "billing_alerts")
|
|
self.bq_client = bigquery.Client(project=self.project_id)
|
|
self.publisher = pubsub_v1.PublisherClient()
|
|
self.topic_path = self.publisher.topic_path(self.project_id, self.pubsub_topic_name)
|
|
self.billing_table = f"{self.project_id}.billing_export.gcp_billing_export_v1_*"
|
|
|
|
def detect_anomalies(self):
|
|
"""
|
|
Detects anomalies in billing data.
|
|
"""
|
|
query = f"""
|
|
WITH
|
|
cost_last_7_days AS (
|
|
SELECT
|
|
service.description as service,
|
|
SUM(cost) as total_cost
|
|
FROM `{self.billing_table}`
|
|
WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 8 DAY)
|
|
AND _PARTITIONTIME < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
|
|
GROUP BY 1
|
|
),
|
|
cost_last_1_day AS (
|
|
SELECT
|
|
service.description as service,
|
|
SUM(cost) as total_cost
|
|
FROM `{self.billing_table}`
|
|
WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
|
|
GROUP BY 1
|
|
)
|
|
SELECT
|
|
c1.service,
|
|
c1.total_cost as today_cost,
|
|
c7.total_cost / 7 as avg_7day,
|
|
c1.total_cost / (c7.total_cost / 7) as ratio
|
|
FROM cost_last_1_day c1
|
|
JOIN cost_last_7_days c7 ON c1.service = c7.service
|
|
WHERE c1.total_cost > 2 * (c7.total_cost / 7)
|
|
"""
|
|
query_job = self.bq_client.query(query)
|
|
results = query_job.result()
|
|
|
|
anomalies = []
|
|
for row in results:
|
|
anomalies.append({
|
|
"service": row.service,
|
|
"today_cost": row.today_cost,
|
|
"avg_7day": row.avg_7day,
|
|
"ratio": row.ratio
|
|
})
|
|
|
|
if anomalies:
|
|
self.publish_alerts(anomalies)
|
|
|
|
return {"anomalies": anomalies}
|
|
|
|
def publish_alerts(self, anomalies):
|
|
"""
|
|
Publishes anomaly alerts to a Pub/Sub topic.
|
|
"""
|
|
for anomaly in anomalies:
|
|
message_data = {
|
|
"service": anomaly["service"],
|
|
"today_cost": anomaly["today_cost"],
|
|
"avg_7day": anomaly["avg_7day"],
|
|
"ratio": anomaly["ratio"],
|
|
"timestamp": datetime.datetime.now().isoformat()
|
|
}
|
|
message_bytes = json.dumps(message_data).encode("utf-8")
|
|
self.publisher.publish(self.topic_path, data=message_bytes)
|
|
|
|
if __name__ == '__main__':
|
|
detector = AnomalyDetector()
|
|
print("Anomalies:", detector.detect_anomalies())
|