import boto3 import datetime from datetime import date from pprint import pprint import calendar import csv class AccountReport: regions = ['us-east-1','eu-west-1'] def __init__(self, profile: str): self.profile = profile self.traffic_cost = 0 self.traffic_amount = 0 self.hours_cost = 0 self.nat_gw_count = 0 self.vpc_with_nat_gw_count = 0 self.year_month = '' self.year = 0 self.month = 0 def vpc_stats(self) -> None: for region in self.regions: session = boto3.Session(profile_name=self.profile,region_name=region) ec2 = session.client('ec2') try: nat_ids = [x['NatGatewayId'] for x in ec2.describe_nat_gateways()['NatGateways'] ] vpc_ids = list(set([x['VpcId'] for x in ec2.describe_nat_gateways()['NatGateways']])) except: nat_ids = [] vpc_ids = [] self.nat_gw_count += len(nat_ids) self.vpc_with_nat_gw_count += len(vpc_ids) def ce_stats(self, month, year) -> None: session = boto3.Session(profile_name=self.profile,region_name=self.regions[0]) client = session.client('ce') endTime = datetime.datetime(year, month + 1, 1, 0, 00) startTime = datetime.datetime(year, month, 1, 0, 00) endDateStr = endTime.strftime("%Y-%m-%d") startDateStr = startTime.strftime("%Y-%m-%d") # Request parameters timePeriod = { 'Start': startDateStr, 'End': endDateStr } granularity = 'MONTHLY' bytes_filter = { 'Dimensions': { 'Key': 'USAGE_TYPE', 'Values': ['NatGateway-Bytes'] }} bytes_metrics = ['NetAmortizedCost', 'UsageQuantity'] hours_filter = { 'Dimensions': { 'Key': 'USAGE_TYPE', 'Values': ['NatGateway-Hours'] }} hours_metrics = ['NetAmortizedCost'] bytes_response = client.get_cost_and_usage( TimePeriod=timePeriod, #GroupBy=groupBy, Granularity=granularity, Filter=bytes_filter, Metrics=bytes_metrics, ) hours_response = client.get_cost_and_usage( TimePeriod=timePeriod, #GroupBy=groupBy, Granularity=granularity, Filter=hours_filter, Metrics=hours_metrics, ) self.traffic_cost = float(bytes_response['ResultsByTime'][0]['Total']['NetAmortizedCost']['Amount']) self.traffic_amount = float(bytes_response['ResultsByTime'][0]['Total']['UsageQuantity']['Amount']) self.hours_cost = float(hours_response['ResultsByTime'][0]['Total']['NetAmortizedCost']['Amount']) self.year_month = f"{year}-{month:02d}" self.month = month self.year = year def print(self): print(f"AWS Account alias: {self.profile}\n" f"Year-month: {self.year_month}\n" f"Nat traffic cost: {self.traffic_cost}\n" f"Nat traffic amount (GB): {self.traffic_amount}\n" f"Nat hours running cost: {self.hours_cost}\n" f"Vpc with nat gateways count {self.vpc_with_nat_gw_count}\n" f"Nat gateways count: {self.nat_gw_count}\n") def to_dict(self) -> dict: return { "Account alias" : self.profile, "Year/Month" : self.year_month, "NAT traffic cost" : self.traffic_cost, "NAT traffic amount (GB)" : self.traffic_amount, "NAT hours cost" : self.hours_cost, "NAT gateways count" : self.nat_gw_count, "VPCs with NAT gateways count" : self.vpc_with_nat_gw_count, } class PredictedCost: az = 6 # us-east-1 fw_traffic_price = 0.065 fw_hour_price = 0.395 tgw_traffic_price = 0.02 tgw_hour_price = 0.05 def __init__(self, report: AccountReport) -> None: month_days = calendar.monthrange(report.year, report.month)[1] self.report = report self.fw_cost = self.fw_traffic_price * report.traffic_amount self.fw_additional_cost = self.fw_cost - report.traffic_cost self.fw_hours_cost = self.fw_hour_price * 24 * month_days * self.az self.fw_hours_additional_cost = self.fw_hours_cost - report.hours_cost self.tgw_cost = self.tgw_traffic_price * report.traffic_amount self.tgw_hours_cost = self.tgw_hour_price * 24 * month_days * (report.vpc_with_nat_gw_count + 1) def print(self) -> None: print(f"Firewall cost:\n" f"- traffic: {self.fw_cost}\n" f"- traffic with nat traffic cost excluded: {self.fw_additional_cost}\n" f"- hours running cost: {self.fw_hours_cost}\n" f"- hours running cost with nat running cost excluded: {self.fw_hours_additional_cost}\n" f"Transit gateway cost:\n" f"- traffic: {self.tgw_cost}\n" f"- hours running cost: {self.tgw_hours_cost}\n") def to_dict(self) -> None: return { "Year/Month" : self.report.year_month, "NAT traffic cost" : self.report.traffic_cost, "NAT traffic amount (GB)" : self.report.traffic_amount, "NAT hours cost" : self.report.hours_cost, "NAT gateways count" : self.report.nat_gw_count, "VPCs with NAT gateways count": self.report.vpc_with_nat_gw_count, "Firewall traffic full cost": self.fw_cost, "Firewall traffic additional cost compared to NAT": self.fw_additional_cost, "Firewall hours running full cost": self.fw_hours_cost, "Firewall hours running additional cost compared to NAT": self.fw_hours_additional_cost, "Transit gateway traffic cost": self.tgw_cost, "Transit gateway endpoint running cost": self.tgw_hours_cost, } def account_yearly_report(profile, year): reports = [] today = date.today() if year != today.year: month_range = range(1,13) else: month_range = range(1,today.month) for month in month_range: report = AccountReport(profile) report.vpc_stats() report.ce_stats(month, year) reports.append(report.to_dict()) return reports def monthly_summary(profiles, year, month) -> AccountReport: summaryReport = AccountReport('') summaryReport.year_month = f"{year}-{month:02d}" summaryReport.month = month summaryReport.year = year for profile in profiles: report = AccountReport(profile) report.vpc_stats() report.ce_stats(month, year) summaryReport.traffic_cost += report.traffic_cost summaryReport.traffic_amount += report.traffic_amount summaryReport.hours_cost += report.hours_cost summaryReport.nat_gw_count += report.nat_gw_count summaryReport.vpc_with_nat_gw_count += report.vpc_with_nat_gw_count return summaryReport def monthly_additional_cost(profiles, year, month ): report = monthly_summary(profiles, year, month ) #report.print() additional_cost = PredictedCost(report) #additional_cost.print() return additional_cost.to_dict() if __name__ == "__main__": dev_profiles = [ "gdb-apollo-dev", "gdb-artistapp-dev", "gdb-core-dev", "gdb-databricks-dev", "gdb-datasci-dev", "gdb-delphi-dev", "gdb-infra-dev", "gdb-mct-dev", "dna-product-hub", "gdb-smu-dev", "gdb-whitelist-dev", ] prod_profiles = [ "gdb-apollo-prod", "gdb-artistapp-prod", "gdb-core-prod", "gdb-databricks-prod", "gdb-delphi-prod", "gdb-infra-prod", "gdb-mct-prod", "gdb-smu-prod", "gdb-whitelist-prod", ] all_profiles = [ "gdb-apollo-dev", "gdb-artistapp-dev", "gdb-core-dev", "gdb-databricks-dev", "gdb-datasci-dev", "gdb-delphi-dev", "gdb-infra-dev", "gdb-mct-dev", "dna-product-hub", "gdb-smu-dev", "gdb-whitelist-dev", "gdb-apollo-prod", "gdb-artistapp-prod", "gdb-core-prod", "gdb-databricks-prod", "gdb-delphi-prod", "gdb-infra-prod", "gdb-mct-prod", "gdb-smu-prod", "gdb-whitelist-prod", "gdb-whitelist-legacy", ] #print("\nDev accs") #monthly_additional_cost(dev_profiles, 2022, 2) #print("\nProd accs") #monthly_additional_cost(prod_profiles, 2022, 2) per_account_report = [] for profile in all_profiles: per_account_report.extend( account_yearly_report(profile, 2022)) with open('monthly_stats_per_account.csv', 'w') as f: w = csv.DictWriter(f, fieldnames=list(per_account_report[0].keys())) w.writeheader() w.writerows(per_account_report) dev_cost_assesment = [] prod_cost_assesment = [] for month in range(1, 8): dev_cost_assesment.append(monthly_additional_cost(dev_profiles, 2022, month)) prod_cost_assesment.append(monthly_additional_cost(prod_profiles, 2022, month)) with open('dev_summary_assesment.csv', 'w') as f: w = csv.DictWriter(f, fieldnames=list(dev_cost_assesment[0].keys())) w.writeheader() w.writerows(dev_cost_assesment) with open('prod_summary_assesment.csv', 'w') as f: w = csv.DictWriter(f, fieldnames=list(prod_cost_assesment[0].keys())) w.writeheader() w.writerows(prod_cost_assesment) #pprint(f"{nat_gateways_count} across all accounts")