Upload finops/finops_scanner.py with huggingface_hub
Browse files- finops/finops_scanner.py +88 -0
finops/finops_scanner.py
ADDED
|
@@ -0,0 +1,88 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
#!/usr/bin/env python3
|
| 2 |
+
"""FinOps Cloud Cost Scanner - Detect Waste & Recommend Savings."""
|
| 3 |
+
|
| 4 |
+
import json, subprocess
|
| 5 |
+
from datetime import datetime
|
| 6 |
+
from pathlib import Path
|
| 7 |
+
from typing import Dict, List
|
| 8 |
+
|
| 9 |
+
class FinOpsScanner:
|
| 10 |
+
def __init__(self, output_dir: str = "./finops-reports"):
|
| 11 |
+
self.output_dir = Path(output_dir)
|
| 12 |
+
self.output_dir.mkdir(parents=True, exist_ok=True)
|
| 13 |
+
self.findings: List[Dict] = []
|
| 14 |
+
self.total_potential_savings = 0.0
|
| 15 |
+
|
| 16 |
+
def _aws_cmd(self, service: str, cmd: str) -> str:
|
| 17 |
+
full_cmd = f"aws {service} {cmd} --output json --region us-east-1"
|
| 18 |
+
try:
|
| 19 |
+
result = subprocess.run(full_cmd, shell=True, capture_output=True, text=True, timeout=60)
|
| 20 |
+
return result.stdout if result.returncode == 0 else "[]"
|
| 21 |
+
except Exception:
|
| 22 |
+
return "[]"
|
| 23 |
+
|
| 24 |
+
def scan_unused_volumes(self) -> List[Dict]:
|
| 25 |
+
output = self._aws_cmd("ec2", "describe-volumes --filters Name=status,Values=available")
|
| 26 |
+
try:
|
| 27 |
+
volumes = json.loads(output).get("Volumes", [])
|
| 28 |
+
for v in volumes:
|
| 29 |
+
size_gb = v["Size"]
|
| 30 |
+
savings = size_gb * 0.08
|
| 31 |
+
self.findings.append({
|
| 32 |
+
"id": f"FINOPS-001-{v['VolumeId']}",
|
| 33 |
+
"type": "unused-ebs",
|
| 34 |
+
"size_gb": size_gb,
|
| 35 |
+
"monthly_savings": round(savings, 2),
|
| 36 |
+
"action": f"Delete or snapshot {v['VolumeId']}",
|
| 37 |
+
})
|
| 38 |
+
self.total_potential_savings += savings
|
| 39 |
+
return self.findings
|
| 40 |
+
except json.JSONDecodeError:
|
| 41 |
+
return []
|
| 42 |
+
|
| 43 |
+
def scan_idle_eips(self) -> List[Dict]:
|
| 44 |
+
output = self._aws_cmd("ec2", "describe-addresses")
|
| 45 |
+
try:
|
| 46 |
+
for addr in json.loads(output).get("Addresses", []):
|
| 47 |
+
if "AssociationId" not in addr:
|
| 48 |
+
self.findings.append({
|
| 49 |
+
"id": f"FINOPS-003-{addr['PublicIp']}",
|
| 50 |
+
"type": "unattached-eip",
|
| 51 |
+
"monthly_savings": 3.60,
|
| 52 |
+
"action": f"Release EIP {addr['PublicIp']}",
|
| 53 |
+
})
|
| 54 |
+
self.total_potential_savings += 3.60
|
| 55 |
+
return self.findings
|
| 56 |
+
except json.JSONDecodeError:
|
| 57 |
+
return []
|
| 58 |
+
|
| 59 |
+
def generate_report(self) -> str:
|
| 60 |
+
report = {
|
| 61 |
+
"timestamp": datetime.utcnow().isoformat() + "Z",
|
| 62 |
+
"total_potential_monthly_savings": round(self.total_potential_savings, 2),
|
| 63 |
+
"findings_count": len(self.findings),
|
| 64 |
+
"findings": self.findings,
|
| 65 |
+
"recommendations": [
|
| 66 |
+
{"priority": 1, "action": "Schedule non-prod off-hours", "savings": "65% non-prod"},
|
| 67 |
+
{"priority": 2, "action": "Purchase RIs for EKS nodes", "savings": "30-40% compute"},
|
| 68 |
+
{"priority": 3, "action": "Enable S3 Intelligent-Tiering", "savings": "40-60% storage"},
|
| 69 |
+
{"priority": 4, "action": "Use SPOT for ML training", "savings": "70-90% GPU"},
|
| 70 |
+
{"priority": 5, "action": "Rightsize K8s workloads", "savings": "30-50% cluster"},
|
| 71 |
+
],
|
| 72 |
+
}
|
| 73 |
+
report_path = self.output_dir / f"finops-{datetime.now().strftime('%Y%m%d')}.json"
|
| 74 |
+
with open(report_path, "w") as f:
|
| 75 |
+
json.dump(report, f, indent=2, default=str)
|
| 76 |
+
print(f"
|
| 77 |
+
{'='*60}")
|
| 78 |
+
print(f"FINOPS SCAN: {len(self.findings)} findings | ${self.total_potential_savings:,.2f}/mo potential savings")
|
| 79 |
+
print(f"{'='*60}")
|
| 80 |
+
return str(report_path)
|
| 81 |
+
|
| 82 |
+
def run_full_scan(self):
|
| 83 |
+
self.scan_unused_volumes()
|
| 84 |
+
self.scan_idle_eips()
|
| 85 |
+
return self.generate_report()
|
| 86 |
+
|
| 87 |
+
if __name__ == "__main__":
|
| 88 |
+
FinOpsScanner().run_full_scan()
|