diff --git a/policies/idle_kusto_cluster.sql b/policies/idle_kusto_cluster.sql new file mode 100644 index 0000000..7803cec --- /dev/null +++ b/policies/idle_kusto_cluster.sql @@ -0,0 +1,18 @@ +-- Real check: Azure Data Explorer cluster with zero ingestion and zero +-- queries over the lookback window while its reserved instance count stays +-- active and billed hourly. This is a genuine reserved-capacity cost pattern +-- rather than a transient idle condition. +SELECT + 'azure' AS provider, + resource_id, + 'Azure Data Explorer Cluster' AS service_name, + ROUND(instance_count * hourly_price_per_instance * 730, 2) AS billed_cost, + resource_name, + sku, + instance_count, + total_ingestion_mb, + total_queries, + lookback_days, + 'cluster with 0 ingestion and 0 queries over ' || lookback_days || ' days (' || sku || ')' AS evidence_reason +FROM fact_idle_kusto_cluster +WHERE total_ingestion_mb = 0 AND total_queries = 0; diff --git a/src/cloudcost/engine/runner.py b/src/cloudcost/engine/runner.py index 6bfd6f1..276b687 100644 --- a/src/cloudcost/engine/runner.py +++ b/src/cloudcost/engine/runner.py @@ -61,6 +61,7 @@ import cloudcost.sources.azure.idle_ai_search import cloudcost.sources.azure.idle_aml_compute_instance import cloudcost.sources.azure.idle_aml_compute_cluster + import cloudcost.sources.azure.idle_kusto_cluster import cloudcost.sources.oci.billing import cloudcost.sources.alibaba.billing except ImportError as e: diff --git a/src/cloudcost/sources/azure/idle_kusto_cluster.py b/src/cloudcost/sources/azure/idle_kusto_cluster.py new file mode 100644 index 0000000..47a659f --- /dev/null +++ b/src/cloudcost/sources/azure/idle_kusto_cluster.py @@ -0,0 +1,201 @@ +import json +import subprocess +from datetime import datetime, timedelta, timezone +from typing import Any + +import pyarrow as pa + +from cloudcost.core.registry import registry + + +def _normalize_region(location: str) -> str: + return location.lower().replace(" ", "") + + +def _fetch_hourly_price(sku_name: str, region: str) -> float: + region = _normalize_region(region) + filter_str = ( + f"armRegionName eq '{region}' and armSkuName eq '{sku_name}' " + f"and priceType eq 'Consumption' and contains(productName, 'Azure Data Explorer') eq true" + ) + + try: + raw = subprocess.run( + [ + "curl", + "-s", + "-G", + "https://prices.azure.com/api/retail/prices", + "--data-urlencode", + f"$filter={filter_str}", + ], + capture_output=True, + text=True, + check=True, + timeout=15, + ).stdout + data = json.loads(raw) + items = [ + item + for item in data.get("Items", []) + if item.get("unitOfMeasure") == "1 Hour" + and item.get("skuName") == sku_name + ] + return float(items[0]["retailPrice"]) if items else 0.0 + except Exception: + return 0.0 + + +# Real check: Azure Data Explorer clusters reserve fixed instance counts +# and are billed hourly even when ingestion and query volume is near zero. +# The check is based on live cluster inventory plus zero-ingestion and zero- +# query totals over the lookback window. +@registry.register_source("azure.idle_kusto_cluster") +class AzureIdleKustoClusterSource: + def __init__(self, config: dict): + self.resource_group = config.get("resource_group") + self.lookback_days = config.get("lookback_days", 7) + + if not self.resource_group: + raise ValueError("azure.idle_kusto_cluster requires 'resource_group' in config") + + def extract(self, context: Any = None) -> pa.Table: + clusters_raw = subprocess.run( + [ + "az", + "kusto", + "cluster", + "list", + "--resource-group", + self.resource_group, + "--query", + "[].{id:id,name:name,sku:sku,location:location}", + "-o", + "json", + ], + capture_output=True, + text=True, + check=True, + ).stdout + clusters = json.loads(clusters_raw) + + end_time = datetime.now(timezone.utc) + start_time = end_time - timedelta(days=self.lookback_days) + + rows = [] + for cluster in clusters: + sku = cluster.get("sku") or {} + if isinstance(sku, dict): + sku_name = sku.get("name") or sku.get("skuName") or "unknown" + instance_count = sku.get("capacity") or 1 + else: + sku_name = sku + instance_count = 1 + + if sku_name == "unknown": + continue + + ingestion_raw = subprocess.run( + [ + "az", + "monitor", + "metrics", + "list", + "--resource", + cluster["id"], + "--metric", + "IngestionVolumeMB", + "--aggregation", + "Total", + "--interval", + "PT1H", + "--start-time", + start_time.strftime("%Y-%m-%dT%H:%M:%SZ"), + "--end-time", + end_time.strftime("%Y-%m-%dT%H:%M:%SZ"), + ], + capture_output=True, + text=True, + check=True, + ).stdout + ingestion_parsed = json.loads(ingestion_raw) + + total_ingestion_mb = 0.0 + for timeseries in ingestion_parsed.get("value", []): + for series in timeseries.get("timeseries", []): + for point in series.get("data", []): + total_ingestion_mb += point.get("total") or 0.0 + + query_raw = subprocess.run( + [ + "az", + "monitor", + "metrics", + "list", + "--resource", + cluster["id"], + "--metric", + "QueryCount", + "--aggregation", + "Total", + "--interval", + "PT1H", + "--start-time", + start_time.strftime("%Y-%m-%dT%H:%M:%SZ"), + "--end-time", + end_time.strftime("%Y-%m-%dT%H:%M:%SZ"), + ], + capture_output=True, + text=True, + check=True, + ).stdout + query_parsed = json.loads(query_raw) + + total_queries = 0.0 + for timeseries in query_parsed.get("value", []): + for series in timeseries.get("timeseries", []): + for point in series.get("data", []): + total_queries += point.get("total") or 0.0 + + rows.append( + { + "resource_id": cluster["id"].lower(), + "resource_name": cluster["name"], + "sku": sku_name, + "instance_count": instance_count, + "total_ingestion_mb": total_ingestion_mb, + "total_queries": total_queries, + "hourly_price_per_instance": _fetch_hourly_price( + sku_name, + cluster.get("location", "eastus"), + ), + "lookback_days": self.lookback_days, + } + ) + + if not rows: + return pa.table( + { + "resource_id": pa.array([], type=pa.string()), + "resource_name": pa.array([], type=pa.string()), + "sku": pa.array([], type=pa.string()), + "instance_count": pa.array([], type=pa.int64()), + "total_ingestion_mb": pa.array([], type=pa.float64()), + "total_queries": pa.array([], type=pa.float64()), + "hourly_price_per_instance": pa.array([], type=pa.float64()), + "lookback_days": pa.array([], type=pa.int64()), + } + ) + + return pa.table( + { + "resource_id": [r["resource_id"] for r in rows], + "resource_name": [r["resource_name"] for r in rows], + "sku": [r["sku"] for r in rows], + "instance_count": [r["instance_count"] for r in rows], + "total_ingestion_mb": [r["total_ingestion_mb"] for r in rows], + "total_queries": [r["total_queries"] for r in rows], + "hourly_price_per_instance": [r["hourly_price_per_instance"] for r in rows], + "lookback_days": [r["lookback_days"] for r in rows], + } + ) diff --git a/tests/policies/test_idle_kusto_cluster.py b/tests/policies/test_idle_kusto_cluster.py new file mode 100644 index 0000000..16c7ed4 --- /dev/null +++ b/tests/policies/test_idle_kusto_cluster.py @@ -0,0 +1,47 @@ +import pyarrow as pa + + +def kusto_rows(*rows: tuple[str, str, int, float, float, float]) -> pa.Table: + return pa.table( + { + "resource_id": pa.array([r[0] for r in rows], type=pa.string()), + "resource_name": pa.array([f"kusto-{r[0]}" for r in rows], type=pa.string()), + "sku": pa.array([r[1] for r in rows], type=pa.string()), + "instance_count": pa.array([r[2] for r in rows], type=pa.int64()), + "total_ingestion_mb": pa.array([r[3] for r in rows], type=pa.float64()), + "total_queries": pa.array([r[4] for r in rows], type=pa.float64()), + "hourly_price_per_instance": pa.array([r[5] for r in rows], type=pa.float64()), + "lookback_days": pa.array([7 for _ in rows], type=pa.int64()), + } + ) + + +def test_kusto_cluster_with_zero_ingestion_and_queries_is_reported(run_policy) -> None: + findings = run_policy( + "idle_kusto_cluster", + { + "fact_idle_kusto_cluster": kusto_rows( + ("cluster1", "Standard_E2a_v4", 2, 0.0, 0.0, 0.12) + ) + }, + ) + + assert len(findings) == 1 + finding = findings[0] + assert finding.provider == "azure" + assert finding.service_name == "Azure Data Explorer Cluster" + assert finding.estimated_impact == 175.2 + assert finding.evidence["evidence_reason"] == "cluster with 0 ingestion and 0 queries over 7 days (Standard_E2a_v4)" + + +def test_kusto_cluster_with_activity_is_not_reported(run_policy) -> None: + findings = run_policy( + "idle_kusto_cluster", + { + "fact_idle_kusto_cluster": kusto_rows( + ("cluster1", "Standard_E2a_v4", 2, 150.0, 12.0, 0.12) + ) + }, + ) + + assert findings == []