-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathec2_package.py
87 lines (75 loc) · 3.13 KB
/
ec2_package.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
from boto3 import Session
from datetime import datetime, timedelta
from decimal import Decimal
def get_regions(session: Session, region='us-east-1') -> list:
client = session.client('ec2', region_name=region)
describe_args = {
'AllRegions': False
}
return [region['RegionName'] for region in client.describe_regions(**describe_args)['Regions']]
def get_instance_types(session: Session, region: str) -> list:
client = session.client('ec2', region_name=region)
describe_args = {
'MaxResults': 300
}
while True:
describe_result = client.describe_instance_types(**describe_args)
yield from [instance['InstanceType'] for instance in describe_result['InstanceTypes']]
if 'NextToken' not in describe_result:
break
describe_args['NextToken'] = describe_result['NextToken']
def get_availability_zones(session: Session, region: str) -> list:
client = session.client('ec2', region_name=region)
describe_args = {
'AllAvailabilityZones': False
}
return [az['ZoneName'] for az in client.describe_availability_zones(**describe_args)['AvailabilityZones']]
def get_it_az(session: Session, region: str) -> str:
client = session.client('ec2', region_name=region)
describe_args = {
'LocationType': 'availability-zone',
}
while True:
response = client.describe_instance_type_offerings(**describe_args)
for obj in response['InstanceTypeOfferings']:
it, region, az = obj.values()
yield it
if 'NextToken' not in response:
break
describe_args['NextToken'] = response['NextToken']
def get_spot_price(session: Session, region: str, start=None, end=None) -> tuple:
if type(end) is not datetime:
end = datetime.utcnow()
if type(start) is not datetime:
start = end - timedelta(hours=1)
client = session.client('ec2', region)
describe_args = {
# 'AvailabilityZone': az,
# 'InstanceTypes': [it],
# 'ProductDescriptions': ['Linux/UNIX (Amazon VPC)'],
'MaxResults': 300,
'EndTime': end,
'StartTime': start
}
while True:
response = client.describe_spot_price_history(**describe_args)
for obj in response['SpotPriceHistory']:
az, it, os, price, timestamp = obj.values()
yield it, az, os, Decimal(price), timestamp
if not response['NextToken']:
break
describe_args['NextToken'] = response['NextToken']
def store_spot_price(params: tuple) -> str:
buffers, session, region, start, end = params
buffer = dict()
for it, az, os, price, timestamp in get_spot_price(session, region, start, end):
if it not in buffer:
buffer[it] = {az: {os: {'price': price, 'timestamp': timestamp}}}
elif az not in buffer[it]:
buffer[it][az] = {os: {'price': price, 'timestamp': timestamp}}
else:
if os in buffer[it][az] and timestamp < buffer[it][az][os]['timestamp']:
continue
buffer[it][az][os] = {'price': price, 'timestamp': timestamp}
buffers.append(buffer)
return region