tools/burst/offers.py (3701 bytes)
1 """Live offers of the clouds: which machines each one can start now, and at what price. 2 3 Each function reads one provider's API and returns `core.Offer` objects. The 4 controller asks again each time it wants a new worker, so a price drop or new 5 stock at one cloud moves the next worker there. 6 """ 7 8 from __future__ import annotations 9 10 from collections.abc import Callable, Iterable 11 12 from .core import Offer 13 14 # Instance types that can run a worker (4 vCPU, 8 GB), and their vCPUs. Arm 15 # types (c7g, c6g) need an arm64 install that nobody has measured yet: turn 16 # them on with SUBPLZ_BURST_AWS_TYPES after one test book. 17 AWS_VCPUS = { 18 "c6a.xlarge": 4, "c7a.xlarge": 4, "m6a.xlarge": 4, "c6i.xlarge": 4, "c7i.xlarge": 4, 19 "c7g.xlarge": 4, "c6g.xlarge": 4, "m7g.xlarge": 4, 20 } 21 AWS_ARM = {"c7g.xlarge", "c6g.xlarge", "m7g.xlarge"} 22 23 24 def aws_offers( 25 spot_prices: Iterable[dict], 26 *, 27 max_price: float, 28 extra_usd_per_hour: float, 29 extra_usd_per_book: float, 30 capacity: int, 31 ) -> list[Offer]: 32 """Offers from AWS spot price history (DescribeSpotPriceHistory items). 33 34 The newest price of each instance type in each availability zone counts. 35 A price above `max_price` is left out: the spot request would not start. 36 `extra_usd_per_hour` is the public IPv4 address and the disk. 37 """ 38 newest: dict[tuple[str, str], tuple[str, float]] = {} 39 for item in spot_prices: 40 key = (item["InstanceType"], item["AvailabilityZone"]) 41 stamp = str(item["Timestamp"]) 42 if key not in newest or stamp > newest[key][0]: 43 newest[key] = (stamp, float(item["SpotPrice"])) 44 return [ 45 Offer( 46 provider="aws", machine=machine, zone=zone, vcpus=AWS_VCPUS[machine], 47 usd_per_hour=price + extra_usd_per_hour, per_started_hour=False, 48 extra_usd_per_book=extra_usd_per_book, capacity=capacity, 49 ) 50 for (machine, zone), (_, price) in sorted(newest.items()) 51 if machine in AWS_VCPUS and price <= max_price 52 ] 53 54 55 def fetch_aws_spot_prices(ec2, machines: list[str], since) -> list[dict]: 56 """DescribeSpotPriceHistory for Linux, all pages.""" 57 items: list[dict] = [] 58 pages = ec2.get_paginator("describe_spot_price_history").paginate( 59 InstanceTypes=machines, ProductDescriptions=["Linux/UNIX"], StartTime=since, 60 ) 61 for page in pages: 62 items.extend(page["SpotPriceHistory"]) 63 return items 64 65 66 def hetzner_offers( 67 get_json: Callable[[str], dict], 68 *, 69 machines: set[str], 70 usd_per_eur: float, 71 ipv4_eur_per_hour: float, 72 capacity: int, 73 ) -> list[Offer]: 74 """Offers from the Hetzner Cloud API: the server types for sale in each 75 location, at the hourly net price, where a datacenter has them in stock.""" 76 server_types = get_json("/server_types?per_page=50")["server_types"] 77 in_stock = { 78 (dc["location"]["name"], type_id) 79 for dc in get_json("/datacenters?per_page=50")["datacenters"] 80 for type_id in dc["server_types"]["available"] 81 } 82 offers = [] 83 for st in server_types: 84 if st["name"] not in machines or st.get("architecture") != "x86" or st.get("deprecation"): 85 continue 86 for price in st["prices"]: 87 if (price["location"], st["id"]) not in in_stock: 88 continue 89 eur = float(price["price_hourly"]["net"]) + ipv4_eur_per_hour 90 offers.append(Offer( 91 provider="hetzner", machine=st["name"], zone=price["location"], 92 vcpus=int(st["cores"]), usd_per_hour=eur * usd_per_eur, 93 per_started_hour=True, capacity=capacity, 94 )) 95 return sorted(offers, key=lambda o: (o.machine, o.zone))