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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
import os
from datetime import datetime, timezone
import boto3
import requests
from botocore.exceptions import BotoCoreError, ClientError
from dotenv import load_dotenv
load_dotenv()
def find_matching_asg_instance():
asg_name = os.getenv("ASG_NAME", "factorio-asg")
ec2 = boto3.client("ec2")
try:
resp = ec2.describe_instances(
Filters=[
{
"Name": "tag:aws:autoscaling:groupName",
"Values": [asg_name],
},
{
"Name": "instance-state-name",
"Values": ["running"],
},
]
)
except (BotoCoreError, ClientError) as exc:
return {"error": f"Failed to query EC2: {exc}"}
instances = []
for reservation in resp.get("Reservations", []):
for instance in reservation.get("Instances", []):
instances.append(
{
"id": instance.get("InstanceId"),
"type": instance.get("InstanceType"),
"az": instance.get("Placement", {}).get("AvailabilityZone"),
}
)
if not instances:
return None
return {"asg_name": asg_name, "instances": [instances[0]]}
def get_current_spot_price(instance_type: str, az: str):
ec2 = boto3.client("ec2")
try:
resp = ec2.describe_spot_price_history(
InstanceTypes=[instance_type],
ProductDescriptions=["Linux/UNIX"],
AvailabilityZone=az,
StartTime=datetime.now(timezone.utc),
MaxResults=1,
)
except (BotoCoreError, ClientError) as exc:
return {"error": f"Failed to fetch spot history: {exc}"}
if not resp.get("SpotPriceHistory"):
return None
return resp["SpotPriceHistory"][0]
def fetch_instance_pricing(instance_type: str, az: str):
url = f"https://go.runs-on.com/api/instances/{instance_type}"
try:
resp = requests.get(
url,
params={"az": az, "platform": "Linux/UNIX"},
timeout=10,
)
resp.raise_for_status()
return resp.json()
except requests.RequestException as exc:
return {"error": str(exc), "instance_type": instance_type, "az": az}
def _normalize_pricing_results(results: list[dict], fallback_az: str) -> list[dict]:
history = []
for item in results:
timestamp = item.get("timestamp")
spot_price = item.get("spotPrice")
if not timestamp or spot_price is None:
continue
normalized_item = {
"timestamp": str(timestamp),
"spotPrice": float(spot_price),
"onDemandPrice": float(item.get("onDemandPrice")) if item.get("onDemandPrice") is not None else None,
"productDescription": item.get("platform"),
"availabilityZone": item.get("az") or fallback_az,
}
history.append(normalized_item)
history.sort(key=lambda row: row["timestamp"])
return history
def get_asg_spot_pricing():
instance_data = find_matching_asg_instance()
if instance_data is None:
return {"error": "No running instances found for ASG"}
if instance_data.get("error"):
return instance_data
instance = instance_data["instances"][0]
pricing_data = fetch_instance_pricing(instance["type"], instance["az"])
if pricing_data.get("error"):
return pricing_data
results = pricing_data.get("results")
if not isinstance(results, list) or not results:
return {
"error": "Pricing API returned no historical points",
"instance_type": instance["type"],
"az": instance["az"],
}
history = _normalize_pricing_results(results, instance["az"])
if not history:
return {
"error": "Pricing API points could not be normalized",
"instance_type": instance["type"],
"az": instance["az"],
}
latest = history[-1]
return {
"asg_name": instance_data.get("asg_name"),
"instance": instance,
"latest": latest,
"history": history,
}
|
