Mock AWS Spot Instance Interruption Handler in Python
A Python class that simulates AWS Spot instance interruption checks, handling the 10% chance of termination, logging state-saving, and storing notice details.
Python code
45 linesimport time
import random
class SpotInstanceHandler:
def __init__(self, instance_id):
self.instance_id = instance_id
self.interruption_notices = []
def start(self):
print(f"Spot instance {self.instance_id} started")
def check_interruption(self):
# Simulate random interruption check (10% chance)
if random.random() < 0.1:
return True
return False
def handle_interruption(self):
notice = {
"instance_id": self.instance_id,
"time": time.time(),
"action": "terminate"
}
self.interruption_notices.append(notice)
print(f"Interruption detected! Saving state for {self.instance_id}")
return notice
def process(self):
self.start()
for _ in range(5):
if self.check_interruption():
return self.handle_interruption()
print(f"Checking interruption for {self.instance_id}...")
time.sleep(0.5)
print(f"No interruption detected for {self.instance_id}")
return None
if __name__ == "__main__":
handler = SpotInstanceHandler("i-12345")
result = handler.process()
if result:
print(f"Notice stored: {result}")
else:
print("Instance running normally")
Output
Spot instance i-12345 started
Checking interruption for i-12345...
Checking interruption for i-12345...
Checking interruption for i-12345...
Checking interruption for i-12345...
Checking interruption for i-12345...
No interruption detected for i-12345
Instance running normally
Note: With a 10% probability, some runs will instead output:
Spot instance i-12345 started
Checking interruption for i-12345...
Interruption detected! Saving state for i-12345
Notice stored: {'instance_id': 'i-12345', 'time': 1710000000.25, 'action': 'terminate'}
(The exact output varies due to random interruption checks and time()).
How it works
This code simulates the logic an application would implement to gracefully handle AWS Spot instance interruptions. The check_interruption method returns True 10% of the time, mimicking the random nature of Spot instance reclaims. When an interruption is detected, handle_interruption records a notice with the instance ID, a Unix timestamp, and the action to take, then prints a save-state message. The process method runs a loop of up to five checks, sleeping briefly between each to represent a polling interval. If an interruption is found, it returns the notice; otherwise, it confirms normal operation. This pattern lets you test your shutdown and checkpointing logic without waiting for a real AWS event.
Common mistakes
- Using time.time() in real production should be replaced with a monotonic clock for interval measurements.
- Forgetting that the mock uses random.random() – make it deterministic in tests by seeding.
- Assuming this handles real AWS calls; in production you'd use the EC2 metadata service or AWS SDK.
- Not handling the `None` return from `process()` properly, leading to `TypeError` when expecting a dict.
Variations
- Replace the random simulation with a real ec2-metadata HTTP call to the Spot interruption endpoint.
- Use asyncio with `await asyncio.sleep()` for a non-blocking interruption check in async services.
Real-world use cases
- Testing graceful shutdown and checkpointing logic for long-running jobs on Spot instances before deploying to production.
- Simulating Spot interruptions in CI pipelines to verify that containerized workloads cleanly persist state.
- Prototyping an interruption handler that integrates with a workflow engine like AWS Step Functions.
Sponsored
More from Cloud + Python
- Build a URL Shortener Client with Python medium
- Create a Cloud Storage Helper Class in Python easy
- Create a Data Helper Class for Beginners in Python easy
- Cross Account Role Chaining Mock Credentials in Python medium
- Exponential Backoff with Jitter for Cloud API Calls in Python medium
- Generate Mock CloudFormation Stack Events in Python easy
Keep learning
Related tutorials and quizzes for this topic.