Spot & Preemptible Instances Guide — 80-90% Discount on Compute

# Request Spot Instances with a launch specification
aws ec2 request-spot-instances \
  --spot-price "0.05" \
  --instance-count 5 \
  --type "one-time" \
  --launch-specification '{
    "ImageId": "ami-0c55b159cbfafe1f0",
    "InstanceType": "m5.large",
    "Placement": {"AvailabilityZoneGroup": "us-east-1a"},
    "SecurityGroups": [{"GroupId": "sg-12345"}]
  }'

# Use Spot Fleet for diversification
aws ec2 request-spot-fleet \
  --spot-fleet-request-config '{
    "AllocationStrategy": "lowestPrice",
    "TargetCapacity": 20,
    "IamFleetRole": "arn:aws:iam::123456789012:role/spot-fleet-role",
    "LaunchSpecifications": [
      {"InstanceType": "m5.large",  "ImageId": "ami-0c55b159cbfafe1f0", "WeightedCapacity": 1},
      {"InstanceType": "m5.xlarge", "ImageId": "ami-0c55b159cbfafe1f0", "WeightedCapacity": 2},
      {"InstanceType": "c5.large",  "ImageId": "ami-0c55b159cbfafe1f0", "WeightedCapacity": 1},
      {"InstanceType": "r5.large",  "ImageId": "ami-0c55b159cbfafe1f0", "WeightedCapacity": 1}
    ]
  }'

# Monitor spot instance interruptions
aws ec2 describe-spot-instance-requests \
  --query "SpotInstanceRequests[?Status.Code=='instance-terminated-by-price']"

{
    "SpotFleetRequestId": "sfr-abc123",
    "SpotFleetRequestState": "active",
    "TargetCapacity": 20,
    "FulfilledCapacity": 20.0
}

import boto3
import json

ec2 = boto3.client('ec2')
sns = boto3.client('sns')

def lambda_handler(event, context):
    """Handle Spot Instance interruption notice."""
    
    # Parse the interruption notice
    detail = event['detail']
    instance_id = detail['instance-id']
    action = detail['instance-action']  # 'terminate', 'stop', 'hibernate'
    
    print(f"Interruption notice for {instance_id}: {action}")
    
    # Save checkpoint before termination
    instance = ec2.describe_instances(InstanceIds=[instance_id])
    tags = {t['Key']: t['Value'] for t in instance['Reservations'][0]['Instances'][0].get('Tags', [])}
    
    checkpoint_data = {
        'instance_id': instance_id,
        'action': action,
        'tags': tags,
        'interruption_time': event['time']
    }
    
    # Publish to checkpoint queue
    sqs = boto3.client('sqs')
    sqs.send_message(
        QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/spot-checkpoints',
        MessageBody=json.dumps(checkpoint_data)
    )
    
    # Notify ops
    sns.publish(
        TopicArn='arn:aws:sns:us-east-1:123456789012:ops-alerts',
        Subject=f'Spot interruption: {instance_id}',
        Message=json.dumps(checkpoint_data)
    )
    
    return {'instance_id': instance_id, 'action': action, 'checkpoint_saved': True}

# Create a Spot VM with Deallocate eviction policy
az vm create \
  --resource-group batch-rg \
  --name batch-vm-01 \
  --image UbuntuLTS \
  --size Standard_D4s_v3 \
  --priority Spot \
  --eviction-policy Deallocate \
  --max-price -1  # Pay up to on-demand price

# Create a Spot VM Scale Set
az vmss create \
  --resource-group batch-rg \
  --name spot-batch-vmss \
  --image UbuntuLTS \
  --instance-count 0 \
  --vm-sku Standard_D4s_v3 \
  --priority Spot \
  --eviction-policy Deallocate \
  --single-placement-group false \
  --max-price 0.05

# Set auto-scaling for Spot VMSS
az monitor autoscale create \
  --resource-group batch-rg \
  --name spot-autoscale \
  --resource spot-batch-vmss \
  --resource-type Microsoft.Compute/virtualMachineScaleSets \
  --min-count 0 \
  --max-count 20 \
  --count 0

az monitor autoscale rule create \
  --resource-group batch-rg \
  --autoscale-name spot-autoscale \
  --condition "Percentage CPU > 70 avg 5m" \
  --scale out 3

# Create a preemptible VM (max 24 hours)
gcloud compute instances create batch-worker \
  --zone us-central1-a \
  --machine-type n2-standard-8 \
  --preemptible \
  --maintenance-policy TERMINATE

# Create a Spot VM (no 24-hour limit)
gcloud compute instances create spot-worker \
  --zone us-central1-a \
  --machine-type n2-standard-8 \
  --provisioning-model SPOT \
  --instance-termination-action STOP

# Create a preemptible instance template for MIG
gcloud compute instance-templates create preemptible-worker \
  --machine-type n2-standard-4 \
  --preemptible \
  --image-family ubuntu-2204-lts \
  --image-project ubuntu-os-cloud \
  --metadata shutdown-script='#!/bin/bash
    echo "Saving checkpoint..."
    gsutil cp /var/state/current.job gs://dodatech-checkpoints/$(hostname).job
    echo "Checkpoint complete."'

# Create managed instance group with preemptible template
gcloud compute instance-groups managed create preemptible-mig \
  --zone us-central1-a \
  --template preemptible-worker \
  --target-size 0 \
  --max-instances 50

# checkpoint_manager.py — resume interrupted spot workloads
import json
import os
import time
import signal
import subprocess

class CheckpointManager:
    """Save and restore job state for spot instance interruptions."""
    
    def __init__(self, checkpoint_dir: str = "/var/checkpoints"):
        self.checkpoint_dir = checkpoint_dir
        os.makedirs(checkpoint_dir, exist_ok=True)
        signal.signal(signal.SIGTERM, self._handle_shutdown)
        
    def save_checkpoint(self, job_id: str, state: dict):
        """Save job progress to disk."""
        path = os.path.join(self.checkpoint_dir, f"{job_id}.json")
        state['timestamp'] = time.time()
        with open(path, 'w') as f:
            json.dump(state, f)
        print(f"Checkpoint saved: {job_id} at {state['progress']:.1f}%")
        
    def load_checkpoint(self, job_id: str) -> dict:
        """Restore job progress from last checkpoint."""
        path = os.path.join(self.checkpoint_dir, f"{job_id}.json")
        if os.path.exists(path):
            with open(path) as f:
                return json.load(f)
        return {'progress': 0, 'items_processed': []}
    
    def run_with_checkpoints(self, items: list, job_id: str, process_func):
        """Process items with periodic checkpointing."""
        state = self.load_checkpoint(job_id)
        start_index = len(state.get('items_processed', []))
        
        for i, item in enumerate(items[start_index:], start=start_index):
            process_func(item)
            state['items_processed'].append(item.id)
            state['progress'] = ((i + 1) / len(items)) * 100
            
            # Save checkpoint every 50 items
            if (i + 1) % 50 == 0:
                self.save_checkpoint(job_id, state)
        
        self.save_checkpoint(job_id, state)
        print(f"Job {job_id} complete. Processed {len(items)} items.")
    
    def _handle_shutdown(self, signum, frame):
        """Save checkpoint on SIGTERM (spot interruption)."""
        print("Interruption detected. Saving final checkpoint...")
        # The main loop calls save_checkpoint, but ensure last state is written
        exit(0)

# Usage
manager = CheckpointManager()
def process(item):
    time.sleep(0.1)  # Simulate work
    print(f"Processed item {item}")

items = list(range(1000))
manager.run_with_checkpoints(items, "data-export-job", Process)

# EKS: Create a spot-managed node group
eksctl create nodegroup \
  --cluster dodatech-eks \
  --name spot-workers \
  --node-type m5.large \
  --nodes-min 0 \
  --nodes-max 20 \
  --spot \
  --spot-instance-types m5.large,c5.large,r5.large \
  --asg-target-group-arn "arn:aws:elasticloadbalancing:..."

# AKS: Create a spot node pool
az aks nodepool add \
  --resource-group dodatech-rg \
  --cluster-name dodatech-aks \
  --name spotpool \
  --node-vm-size Standard_D4s_v3 \
  --priority Spot \
  --eviction-policy Delete \
  --spot-max-price -1 \
  --enable-cluster-autoscaler \
  --min-count 0 \
  --max-count 10