from calendar import monthrange import csv from datetime import date from datetime import datetime from datetime import timedelta import boto3 STORAGE_PRICES_GB = { 'StandardStorage': 0.01302, 'IntelligentTieringFAStorage': 0.01302, 'IntelligentTieringIAStorage': 0.00775, 'IntelligentTieringAIAStorage': 0.004, 'GIR': 0.004 } REQUEST_COST_STANDARD_PER_1000 = 0.0004 REQUEST_COST_GIR_PER_1000 = 0.01 DATA_RETRIEVAL_COST_GIR_PER_GB = 0.03 GIR_LIFECYCLE_TRANSITION_COST_PER_1000 = 0.02 S3_BUCKET = 'prod-orcd-mezzanine-assets' START_DATE = date(2023, 12, 1) END_DATE = date.today() cloudwatch_client = boto3.client('cloudwatch') s3_client = boto3.client('s3') def get_data_by_storage_class(query_date): """Retrieve the amount of data in either Standard or Intelligent Tiering, broken down by storage class.""" data_by_storage_class = {} for storage_class in ['StandardStorage', 'IntelligentTieringFAStorage', 'IntelligentTieringIAStorage', 'IntelligentTieringAIAStorage']: metric_details = cloudwatch_client.get_metric_data( MetricDataQueries=[ { 'Id': 'test', 'MetricStat': { 'Metric': { 'Namespace': 'AWS/S3', 'MetricName': 'BucketSizeBytes', 'Dimensions': [ { 'Name': 'BucketName', 'Value': S3_BUCKET }, { 'Name': 'StorageType', 'Value': storage_class } ] }, 'Period': 60 * 60 * 24, 'Stat': 'Average' } } ], StartTime=datetime.combine(query_date, datetime.min.time()), EndTime=datetime.combine(query_date, datetime.max.time()), ) if len(metric_details['MetricDataResults'][0]['Values']) == 0: data_by_storage_class[storage_class] = 0 else: data_by_storage_class[storage_class] = metric_details['MetricDataResults'][0]['Values'][0] / (1024 * 1024 * 1024) return data_by_storage_class def load_storage_class_analysis(): """Load the S3 storage class analysis from S3""" s3_object = s3_client.get_object( Bucket='prod-orcd-s3-storage-class-analysis', Key=f'{S3_BUCKET}/{S3_BUCKET}_temp-cost-optimization-analysis.csv' ) data = s3_object['Body'].read().decode('utf-8').splitlines() reader = csv.DictReader(data) return [row for row in reader] def calculate_storage_costs(date, data_by_storage_class): """Calculate the storage costs for the given date, given the amount of data stored in each storage class""" number_of_days_in_month = monthrange(date.year, date.month)[1] return sum(STORAGE_PRICES_GB[storage_class] * data for storage_class, data in data_by_storage_class.items()) / number_of_days_in_month def get_metrics_for_age_ranges(storage_class_analysis, date, *age_ranges): """Retrieve metrics for the given age ranges and date from the storage class analysis""" metrics = {} rows = [ row for row in storage_class_analysis if row['Date'] == date.strftime('%Y-%m-%d') and row['ObjectAge'] in age_ranges and row['StorageClass'] in ['STANDARD', 'INTELLIGENT_TIERING'] ] metrics['StorageGB'] = sum(float(row['Storage_MB'] or '0') for row in rows) / 1024 metrics['GetRequests'] = sum(int(row['GetRequestCount'] or '0') for row in rows) metrics['DataRetrievedGB'] = sum(float(row['DataRetrieved_MB'] or '0') for row in rows) / 1024 metrics['ObjectCount'] = sum(int(row['ObjectCount'] or '0') for row in rows) return metrics def create_csv(results): """Write the results to a CSV""" with open('s3.csv', 'w') as csvfile: writer = csv.DictWriter(csvfile, fieldnames=results[0].keys()) writer.writeheader() for row in results: writer.writerow(row) def main(): results = [] storage_class_analysis = load_storage_class_analysis() date_to_process = START_DATE while date_to_process < END_DATE: print(f'Processing date {date_to_process}') data_by_storage_class = get_data_by_storage_class(date_to_process) metrics_for_data_less_than_15_days = get_metrics_for_age_ranges(storage_class_analysis, date_to_process, '000-014') metrics_for_data_less_than_30_days = get_metrics_for_age_ranges(storage_class_analysis, date_to_process, '000-014', '015-029') metrics_for_data_any_age = get_metrics_for_age_ranges(storage_class_analysis, date_to_process, 'ALL') total_storage_gb = metrics_for_data_any_age['StorageGB'] total_requests = metrics_for_data_any_age['GetRequests'] total_data_retrieved_gb = metrics_for_data_any_age['DataRetrievedGB'] current_storage_costs = calculate_storage_costs(date_to_process, data_by_storage_class) current_request_costs = total_requests * REQUEST_COST_STANDARD_PER_1000 / 1000 current_total_costs = current_storage_costs + current_request_costs # Determine the equivalent costs if all data older than 15 days was in GIR data_by_storage_class_gir_15_days = { 'StandardStorage': metrics_for_data_less_than_15_days['StorageGB'], 'GIR': total_storage_gb - metrics_for_data_less_than_15_days['StorageGB'] } gir_15_day_storage_costs = calculate_storage_costs(date_to_process, data_by_storage_class_gir_15_days) gir_15_day_gir_requests = total_requests - metrics_for_data_less_than_15_days['GetRequests'] gir_15_day_request_costs = ((metrics_for_data_less_than_15_days['GetRequests'] * REQUEST_COST_STANDARD_PER_1000 / 1000) + (gir_15_day_gir_requests * REQUEST_COST_GIR_PER_1000 / 1000)) gir_15_day_gir_data_retrieved = total_data_retrieved_gb - metrics_for_data_less_than_15_days['DataRetrievedGB'] gir_15_day_data_retrieval_costs = DATA_RETRIEVAL_COST_GIR_PER_GB * gir_15_day_gir_data_retrieved gir_15_day_total_costs = gir_15_day_storage_costs + gir_15_day_request_costs + gir_15_day_data_retrieval_costs # Determine the equivalent costs if all data older than 30 days was in GIR data_by_storage_class_gir_30_days = { 'StandardStorage': metrics_for_data_less_than_30_days['StorageGB'], 'GIR': total_storage_gb - metrics_for_data_less_than_30_days['StorageGB'] } gir_30_day_storage_costs = calculate_storage_costs(date_to_process, data_by_storage_class_gir_30_days) gir_30_day_gir_requests = total_requests - metrics_for_data_less_than_30_days['GetRequests'] gir_30_day_request_costs = ((metrics_for_data_less_than_30_days['GetRequests'] * REQUEST_COST_STANDARD_PER_1000 / 1000) + (gir_30_day_gir_requests * REQUEST_COST_GIR_PER_1000 / 1000)) gir_30_day_gir_data_retrieved = total_data_retrieved_gb - metrics_for_data_less_than_30_days['DataRetrievedGB'] gir_30_day_data_retrieval_costs = DATA_RETRIEVAL_COST_GIR_PER_GB * gir_30_day_gir_data_retrieved gir_30_day_total_costs = gir_30_day_storage_costs + gir_30_day_request_costs + gir_30_day_data_retrieval_costs # Estimate upfront transition costs. Storage class analysis doesn't provide object count by age so just estimate based # on transitioning all objects, which will provide a high watermark for the actual costs. upfront_transition_costs = metrics_for_data_any_age['ObjectCount'] * GIR_LIFECYCLE_TRANSITION_COST_PER_1000 / 1000 results_for_date = { 'Date': date_to_process.strftime('%Y-%m-%d'), 'StandardStorageGB': data_by_storage_class['StandardStorage'], 'IntelligentTieringFAStorageGB': data_by_storage_class['IntelligentTieringFAStorage'], 'IntelligentTieringIAStorageGB': data_by_storage_class['IntelligentTieringIAStorage'], 'IntelligentTieringAIAStorageGB': data_by_storage_class['IntelligentTieringAIAStorage'], 'TotalStorageGB': total_storage_gb, 'ProjectedGIRStorage15DayTransition': data_by_storage_class_gir_15_days['GIR'], 'ProjectedGIRStorage30DayTransition': data_by_storage_class_gir_30_days['GIR'], 'CurrentStorageCosts': current_storage_costs, 'ProjectedStorageCosts15DayGIR': gir_15_day_storage_costs, 'ProjectedStorageCosts30DayGIR': gir_30_day_storage_costs, 'TotalRequests': total_requests, 'ProjectedGIRRequests15DayTransition': gir_15_day_gir_requests, 'ProjectedGIRRequests30DayTransition': gir_30_day_gir_requests, 'CurrentRequestCost': current_request_costs, 'ProjectedRequestCosts15DayGIR': gir_15_day_request_costs, 'ProjectedRequestCosts30DayGIR': gir_30_day_request_costs, 'TotalDataRetrievedGB': total_data_retrieved_gb, 'ProjectedGIRDataRetrieved15DayTransition': gir_15_day_gir_data_retrieved, 'ProjectedGIRDataRetrieved30DayTransition': gir_30_day_gir_data_retrieved, 'ProjectedDataRetrievalCosts15DayGIR': gir_15_day_data_retrieval_costs, 'ProjectedDataRetrievalCosts30DayGIR': gir_30_day_data_retrieval_costs, 'CurrentTotalCosts': current_total_costs, 'ProjectedTotalCosts15DayGIR': gir_15_day_total_costs, 'ProjectedTotalCosts30DayGIR': gir_30_day_total_costs, 'UpfrontTransitionCosts': upfront_transition_costs } results.append(results_for_date) date_to_process += timedelta(days=1) create_csv(results) if __name__ == '__main__': main()