Created
May 31, 2020 13:44
-
-
Save Burekasim/6f2eab97725ba5f43227ae77e2f29007 to your computer and use it in GitHub Desktop.
pulls s3 bucket usage per storage engine
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import boto3 | |
from datetime import timedelta, datetime | |
### Logic: | |
# this script will list all aws buckets, and will retrieve each bucket size per storage type, | |
# output will be displayed in mb, if output is 0Mb, the specific storage size is less than 1Mb. | |
# output will only display used storage types | |
### | |
class Aws: | |
def __init__(self, aws_region='us-east-1', profile='default'): | |
self.region = aws_region | |
self.session = boto3.Session(region_name=aws_region, profile_name=profile) | |
def get_bucket_stats(self, bucket_name, bucket_location, storage_type): | |
cw = self.session.client('cloudwatch', region_name=bucket_location) | |
response = cw.get_metric_statistics( | |
Namespace='AWS/S3', MetricName='BucketSizeBytes', | |
StartTime=datetime.utcnow() - timedelta(days=2), | |
EndTime=datetime.utcnow(), Period=86400, | |
Statistics=['Average'], Unit='Bytes', | |
Dimensions=[ | |
{'Name': 'BucketName', 'Value': bucket_name}, | |
{u'Name': 'StorageType', u'Value': storage_type} | |
]) | |
if response: | |
try: | |
ret = response['Datapoints'][0]['Average'] | |
except Exception as e: | |
ret = 0 | |
return ret | |
return 0 | |
def get_bucket_location(self, req_bucket_name): | |
response = self.session.client('s3').get_bucket_location(Bucket=req_bucket_name) | |
bucket_loc = 'us-east-1' | |
if response['LocationConstraint'] is not None: | |
bucket_loc = response['LocationConstraint'] | |
if response['LocationConstraint'] == 'EU': | |
bucket_loc = 'eu-west-1' | |
return bucket_loc | |
if __name__ == '__main__': | |
# us-east-1 - for api calls | |
# default - default profile in ~/.aws/config | |
aws = Aws('us-east-1', 'default') | |
buckets = aws.session.client('s3').list_buckets() | |
if buckets: | |
for bucket_list in buckets['Buckets']: | |
bucket_name = bucket_list['Name'] | |
try: | |
bucket_location = aws.get_bucket_location(bucket_name) | |
except Exception as e: | |
print(str(e)) | |
types = ['GlacierStorage', 'GlacierStorageOverhead', 'IntelligentTieringFAStorage', | |
'IntelligentTieringIAStorage', 'OneZoneIASizeOverhead', 'OneZoneIAStorage', 'ReducedRedundancyStorage', | |
'StandardIASizeOverhead', 'StandardIAStorage', 'StandardStorage'] | |
print(f'region: {bucket_location}, bucket name: {bucket_name}:') | |
bucket_total_size = [] | |
for storage_type in types: | |
bucket_size = aws.get_bucket_stats(bucket_name, bucket_location, storage_type) | |
bucket_size_in_mb = round(bucket_size / 1024 / 1024) | |
if bucket_size: | |
bucket_total_size.append(bucket_size_in_mb) | |
print(f' Storage type: {storage_type}, size in Mb: {bucket_size_in_mb}') | |
if bucket_total_size and len(bucket_total_size) > 1: | |
bucket_total = sum(bucket_total_size) | |
if bucket_total > 1048576: | |
bucket_total = round(bucket_total/1048576, 2) | |
print(f'Bucket total size in \033[1mTb\033[0m: {bucket_total}') | |
elif bucket_total > 1024: | |
bucket_total = round(bucket_total/1024, 2) | |
print(f'Bucket total size in \033[1mGb\033[0m: {bucket_total}') | |
else: | |
print(f'Bucket total size in Mb: {bucket_total}') | |
print('\n') | |
else: | |
print('Couldn\'t retrieve bucket list') |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment