Auto Stop/Start EC2 Instances Using Schedule Tags with Python

pip install boto3 schedule

# Option A — AWS CLI (recommended for local use)
aws configure
# AWS Access Key ID:     AKIA...
# AWS Secret Access Key: xxxxxxxx
# Default region:        ap-south-1
# Default output format: json

# Option B — Environment variables (for CI/CD or Docker)
export AWS_ACCESS_KEY_ID=AKIA...
export AWS_SECRET_ACCESS_KEY=xxxxxxxx
export AWS_DEFAULT_REGION=ap-south-1

# Option C — IAM Role (best for EC2/Lambda — no credentials needed)
# Attach an IAM role with the right permissions to the instance running this script

# Tag an instance via AWS CLI
aws ec2 create-tags \
  --resources i-0abc123def456789 \
  --tags Key=AutoStop,Value=true

# Or tag multiple instances at once
aws ec2 create-tags \
  --resources i-0abc123def456789 i-0def456789abc123 \
  --tags Key=AutoStop,Value=true

# Verify the tag
aws ec2 describe-tags \
  --filters "Name=resource-id,Values=i-0abc123def456789"

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "EC2ScheduleControl",
      "Effect": "Allow",
      "Action": [
        "ec2:DescribeInstances",
        "ec2:DescribeTags",
        "ec2:StartInstances",
        "ec2:StopInstances"
      ],
      "Resource": "*"
    }
  ]
}

# ec2_scheduler.py
"""
EC2 Auto Stop/Start Scheduler
Stops instances tagged AutoStop=true at 8 PM.
Starts them at 8 AM.
"""

import boto3
import schedule
import time
import logging
from datetime import datetime
from botocore.exceptions import ClientError, NoCredentialsError

# ── Logging setup ────────────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  %(levelname)-8s  %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    handlers=[
        logging.StreamHandler(),                    # Print to console
        logging.FileHandler("ec2_scheduler.log"),   # Also write to file
    ],
)
log = logging.getLogger(__name__)

# ── Configuration ─────────────────────────────────────────────────
REGION        = "ap-south-1"   # Change to your AWS region
TAG_KEY       = "AutoStop"     # The tag key we look for
TAG_VALUE     = "true"         # The tag value we look for
STOP_TIME     = "20:00"        # 8 PM — 24-hour format
START_TIME    = "08:00"        # 8 AM — 24-hour format


# ── boto3 client ──────────────────────────────────────────────────
def get_ec2_client():
    """
    Create an EC2 client for the specified region.

    boto3.client() creates a low-level service client.
    - 'ec2'     : the AWS service name
    - region_name: which AWS region to connect to
    Credentials are picked up automatically from:
      1. Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
      2. ~/.aws/credentials file (set by `aws configure`)
      3. IAM role attached to the EC2/Lambda running this script
    """
    return boto3.client("ec2", region_name=REGION)


# ── Helper: find tagged instances ─────────────────────────────────
def get_tagged_instances(ec2, desired_state: str) -> list[dict]:
    """
    Return EC2 instances that have AutoStop=true AND are in the desired_state.

    Parameters
    ----------
    ec2           : boto3 EC2 client
    desired_state : 'running' to find instances to stop,
                    'stopped' to find instances to start

    How describe_instances works
    ----------------------------
    - Filters is a list of dicts with 'Name' and 'Values' keys.
    - 'tag:AutoStop'     matches the tag KEY named AutoStop.
    - 'instance-state-name' filters by the current lifecycle state.
    - The API returns a paginated response — Reservations is the top-level list.
      Each Reservation can contain multiple Instances (from a single launch command).
    """
    try:
        response = ec2.describe_instances(
            Filters=[
                {
                    "Name":   f"tag:{TAG_KEY}",    # Filter by tag key
                    "Values": [TAG_VALUE],          # Tag value must be 'true'
                },
                {
                    "Name":   "instance-state-name",
                    "Values": [desired_state],      # Only running or stopped
                },
            ]
        )
    except NoCredentialsError:
        log.error("AWS credentials not found. Run 'aws configure' or set env vars.")
        return []
    except ClientError as e:
        log.error(f"AWS API error: {e.response['Error']['Message']}")
        return []

    # Flatten the nested Reservations → Instances structure into a flat list
    instances = [
        instance
        for reservation in response["Reservations"]
        for instance in reservation["Instances"]
    ]

    return instances


# ── Helper: get a human-readable instance name ─────────────────────
def get_instance_name(instance: dict) -> str:
    """
    Extract the 'Name' tag value from the instance's Tags list.

    Tags is a list of {'Key': '...', 'Value': '...'} dicts.
    We use next() with a default so it doesn't crash if 'Name' tag is absent.
    """
    tags = instance.get("Tags", [])
    name = next((t["Value"] for t in tags if t["Key"] == "Name"), "Unnamed")
    return name


# ── STOP action ───────────────────────────────────────────────────
def stop_instances():
    """
    Find all running instances tagged AutoStop=true and stop them.
    Called automatically at STOP_TIME (8 PM).
    """
    log.info("=" * 55)
    log.info(f"[STOP JOB] Starting at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

    ec2 = get_ec2_client()
    instances = get_tagged_instances(ec2, desired_state="running")

    if not instances:
        log.info("[STOP JOB] No running instances with AutoStop=true found.")
        return

    # Extract just the instance IDs — that's what start/stop APIs need
    instance_ids = [inst["InstanceId"] for inst in instances]

    log.info(f"[STOP JOB] Found {len(instance_ids)} instance(s) to stop:")
    for inst in instances:
        log.info(f"  - {inst['InstanceId']}  ({get_instance_name(inst)})")

    try:
        """
        ec2.stop_instances() sends a stop signal to each instance.
        - InstanceIds: list of instance ID strings (e.g. ['i-0abc...', 'i-0def...'])
        - The API call is asynchronous — it returns immediately.
          The instance transitions: running → stopping → stopped.
        - DryRun=True can be used to test without actually stopping.
        - StopInstances does NOT terminate (delete) the instance.
          Data on the EBS root volume is preserved.
        """
        response = ec2.stop_instances(InstanceIds=instance_ids)

        # Log the new state reported by AWS for each instance
        for item in response["StoppingInstances"]:
            prev  = item["PreviousState"]["Name"]
            curr  = item["CurrentState"]["Name"]
            iid   = item["InstanceId"]
            log.info(f"  ✓ {iid}: {prev} → {curr}")

        log.info(f"[STOP JOB] Stop signal sent to {len(instance_ids)} instance(s).")

    except ClientError as e:
        error_code = e.response["Error"]["Code"]
        error_msg  = e.response["Error"]["Message"]
        log.error(f"[STOP JOB] Failed to stop instances. Code: {error_code} — {error_msg}")
        # Common errors:
        # UnsupportedOperation: instance store-backed instances can't be stopped
        # IncorrectInstanceState: instance is already stopping/terminated


# ── START action ──────────────────────────────────────────────────
def start_instances():
    """
    Find all stopped instances tagged AutoStop=true and start them.
    Called automatically at START_TIME (8 AM).
    """
    log.info("=" * 55)
    log.info(f"[START JOB] Starting at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

    ec2 = get_ec2_client()
    instances = get_tagged_instances(ec2, desired_state="stopped")

    if not instances:
        log.info("[START JOB] No stopped instances with AutoStop=true found.")
        return

    instance_ids = [inst["InstanceId"] for inst in instances]

    log.info(f"[START JOB] Found {len(instance_ids)} instance(s) to start:")
    for inst in instances:
        log.info(f"  - {inst['InstanceId']}  ({get_instance_name(inst)})")

    try:
        """
        ec2.start_instances() boots stopped EBS-backed instances.
        - InstanceIds: list of instance ID strings.
        - The API call is asynchronous.
          The instance transitions: stopped → pending → running.
        - A new public IP is assigned (unless an Elastic IP is attached).
        - Instance store data is NOT preserved across stop/start cycles.
          EBS data IS preserved.
        """
        response = ec2.start_instances(InstanceIds=instance_ids)

        for item in response["StartingInstances"]:
            prev  = item["PreviousState"]["Name"]
            curr  = item["CurrentState"]["Name"]
            iid   = item["InstanceId"]
            log.info(f"  ✓ {iid}: {prev} → {curr}")

        log.info(f"[START JOB] Start signal sent to {len(instance_ids)} instance(s).")

    except ClientError as e:
        error_code = e.response["Error"]["Code"]
        error_msg  = e.response["Error"]["Message"]
        log.error(f"[START JOB] Failed to start instances. Code: {error_code} — {error_msg}")


# ── Scheduler setup ───────────────────────────────────────────────
def setup_schedule():
    """
    Register stop and start jobs with the schedule library.

    schedule.every().day.at("HH:MM") registers a recurring daily job.
    - Times use 24-hour format.
    - The scheduler runs in the LOCAL timezone of the machine.
      If you need UTC or a specific timezone, use pytz:
          import pytz
          tz = pytz.timezone("Asia/Kolkata")
          now = datetime.now(tz)
    """
    schedule.every().day.at(STOP_TIME).do(stop_instances)
    schedule.every().day.at(START_TIME).do(start_instances)

    log.info(f"Scheduler active — Stop: {STOP_TIME} | Start: {START_TIME}")
    log.info(f"Region: {REGION} | Tag filter: {TAG_KEY}={TAG_VALUE}")
    log.info("Waiting for next scheduled job... (Ctrl+C to exit)")


# ── Entry point ───────────────────────────────────────────────────
if __name__ == "__main__":
    setup_schedule()

    """
    schedule.run_pending() checks if any registered job is due and runs it.
    It does NOT block — it returns immediately if no job is due.

    We wrap it in an infinite loop with time.sleep(60) so we check
    every 60 seconds. This is lightweight (almost no CPU usage while sleeping).

    The loop runs forever until:
    - You press Ctrl+C (raises KeyboardInterrupt)
    - The process is killed (systemd, Docker, etc.)
    """
    try:
        while True:
            schedule.run_pending()
            time.sleep(60)   # Check every minute — fine-grained enough for HH:MM scheduling
    except KeyboardInterrupt:
        log.info("Scheduler stopped by user.")

# Run and watch the logs
python ec2_scheduler.py

# 2025-01-20 07:59:00  INFO     Scheduler active — Stop: 20:00 | Start: 08:00
# 2025-01-20 07:59:00  INFO     Region: ap-south-1 | Tag filter: AutoStop=true
# 2025-01-20 08:00:00  INFO     ═══════════════════════════════════════════════════════
# 2025-01-20 08:00:00  INFO     [START JOB] Starting at 2025-01-20 08:00:00
# 2025-01-20 08:00:00  INFO     Found 3 instance(s) to start:
# 2025-01-20 08:00:00  INFO       - i-0abc123  (dev-api-server)
# 2025-01-20 08:00:00  INFO       - i-0def456  (staging-db)
# 2025-01-20 08:00:00  INFO       - i-0ghi789  (test-worker)
# 2025-01-20 08:00:01  INFO       ✓ i-0abc123: stopped → pending
# 2025-01-20 08:00:01  INFO       ✓ i-0def456: stopped → pending
# 2025-01-20 08:00:01  INFO       ✓ i-0ghi789: stopped → pending

# /etc/systemd/system/ec2-scheduler.service
[Unit]
Description=EC2 Auto Stop/Start Scheduler
After=network.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/opt/ec2-scheduler
ExecStart=/opt/ec2-scheduler/venv/bin/python ec2_scheduler.py
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable ec2-scheduler
sudo systemctl start ec2-scheduler
sudo systemctl status ec2-scheduler
journalctl -u ec2-scheduler -f   # Follow live logs

# Edit crontab
crontab -e

# Add these two lines:
# Stop at 8 PM (20:00) every day
0 20 * * * /usr/bin/python3 /opt/ec2-scheduler/ec2_scheduler_once.py stop >> /var/log/ec2-stop.log 2>&1

# Start at 8 AM (08:00) every day
0 8  * * * /usr/bin/python3 /opt/ec2-scheduler/ec2_scheduler_once.py start >> /var/log/ec2-start.log 2>&1

# ec2_scheduler_once.py — for cron (runs once, stops/starts, then exits)
import sys
from ec2_scheduler import start_instances, stop_instances

if __name__ == "__main__":
    action = sys.argv[1] if len(sys.argv) > 1 else "stop"
    if action == "stop":
        stop_instances()
    elif action == "start":
        start_instances()

# lambda_handler.py — deploy this as a Lambda function
import boto3
from botocore.exceptions import ClientError

TAG_KEY   = "AutoStop"
TAG_VALUE = "true"
REGION    = "ap-south-1"

def lambda_handler(event, context):
    """
    EventBridge triggers this Lambda on a cron schedule.
    event['action'] is set by the EventBridge rule's Input field.
    """
    action = event.get("action", "stop")
    ec2 = boto3.client("ec2", region_name=REGION)

    state = "running" if action == "stop" else "stopped"
    response = ec2.describe_instances(
        Filters=[
            {"Name": f"tag:{TAG_KEY}", "Values": [TAG_VALUE]},
            {"Name": "instance-state-name", "Values": [state]},
        ]
    )

    instance_ids = [
        inst["InstanceId"]
        for r in response["Reservations"]
        for inst in r["Instances"]
    ]

    if not instance_ids:
        return {"status": "no instances found", "action": action}

    if action == "stop":
        ec2.stop_instances(InstanceIds=instance_ids)
    else:
        ec2.start_instances(InstanceIds=instance_ids)

    return {
        "status": "success",
        "action": action,
        "instances": instance_ids,
        "count": len(instance_ids),
    }

# EventBridge rules (AWS Console or CLI)
# Stop rule — every day at 8 PM UTC
aws events put-rule \
  --name ec2-stop-rule \
  --schedule-expression "cron(0 14 * * ? *)" \
  --state ENABLED

# Start rule — every day at 8 AM UTC
aws events put-rule \
  --name ec2-start-rule \
  --schedule-expression "cron(0 2 * * ? *)" \
  --state ENABLED

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  %(levelname)-8s  %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler("ec2_scheduler.log"),
    ],
)

def get_ec2_client():
    return boto3.client("ec2", region_name=REGION)

response = ec2.describe_instances(
    Filters=[
        {"Name": f"tag:{TAG_KEY}", "Values": [TAG_VALUE]},
        {"Name": "instance-state-name", "Values": [desired_state]},
    ]
)

tags = instance.get("Tags", [])
name = next((t["Value"] for t in tags if t["Key"] == "Name"), "Unnamed")

response = ec2.stop_instances(InstanceIds=instance_ids)
for item in response["StoppingInstances"]:
    prev = item["PreviousState"]["Name"]
    curr = item["CurrentState"]["Name"]

response = ec2.start_instances(InstanceIds=instance_ids)
for item in response["StartingInstances"]:
    prev = item["PreviousState"]["Name"]   # "stopped"
    curr = item["CurrentState"]["Name"]    # "pending"

schedule.every().day.at(STOP_TIME).do(stop_instances)
schedule.every().day.at(START_TIME).do(start_instances)

while True:
    schedule.run_pending()
    time.sleep(60)