如何保护 EC2 实例存储（Instance Store）数据不丢失：场景分析与自动化防护实践

aws ec2 modify-instance-attribute \
    --instance-id i-xxx \
    --disable-api-stop

aws ec2 modify-instance-attribute \
    --instance-id i-xxx \
    --disable-api-termination

aws ec2 create-tags \
    --resources i-xxx \
    --tags Key=StopForbidden,Value=true

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "DenyStopAndTerminateProtectedInstances",
            "Effect": "Deny",
            "Action": [
                "ec2:StopInstances",
                "ec2:TerminateInstances"
            ],
            "Resource": "arn:aws:ec2:*:*:instance/*",
            "Condition": {
                "StringEquals": {
                    "ec2:ResourceTag/StopForbidden": "true"
                }
            }
        }
    ]
}

# 已有实例
aws ec2 modify-instance-maintenance-options \ 
    --instance-id i-xxx \
    --auto-recovery disabled

# 新建实例（在 run-instances 中指定）
aws ec2 run-instances \
    --instance-type i3.8xlarge \
    --maintenance-options AutoRecovery=Disabled \
    ...

# 将 Min/Max/Desired 设为相同值，或禁用 scaling policy
aws autoscaling update-auto-scaling-group \
    --auto-scaling-group-name my-instance-store-asg \
    --min-size 3 --max-size 3 --desired-capacity 3

apiVersion: v1
kind: Pod
metadata:
  annotations:
    karpenter.sh/do-not-disrupt: "true"

import boto3
import os
from functools import lru_cache

ec2 = boto3.client('ec2')
sns = boto3.client('sns')
SNS_TOPIC_ARN = os.environ.get('SNS_TOPIC_ARN', '')
PROTECTION_TAG = {'Key': 'StopForbidden', 'Value': 'true'}


@lru_cache(maxsize=256)
def has_instance_store(instance_type: str) -> bool:
    """
    通过 describe_instance_types API 准确判断实例是否支持 Instance Store。
    比硬编码 family 列表更健壮，能自动适配新实例类型。
    """
    try:
        resp = ec2.describe_instance_types(InstanceTypes=[instance_type])
        info = resp['InstanceTypes'][0]
        return info.get('InstanceStorageSupported', False)
    except Exception:
        return False


def enable_protection(instance_id: str) -> dict:
    """为实例启用全套保护"""
    results = {}
    try:
        ec2.modify_instance_attribute(
            InstanceId=instance_id,
            DisableApiStop={'Value': True}
        )
        results['stop_protection'] = 'enabled'
    except Exception as e:
        results['stop_protection'] = f'FAILED: {e}'

    try:
        ec2.modify_instance_attribute(
            InstanceId=instance_id,
            DisableApiTermination={'Value': True}
        )
        results['termination_protection'] = 'enabled'
    except Exception as e:
        results['termination_protection'] = f'FAILED: {e}'

    try:
        ec2.create_tags(
            Resources=[instance_id],
            Tags=[PROTECTION_TAG]
        )
        results['tag'] = 'applied'
    except Exception as e:
        results['tag'] = f'FAILED: {e}'

    try:
        ec2.modify_instance_maintenance_options(
            InstanceId=instance_id,
            AutoRecovery='disabled'
        )
        results['auto_recovery'] = 'disabled'
    except Exception as e:
        results['auto_recovery'] = f'FAILED: {e}'

    return results


def audit_all_instances() -> list:
    """巡检所有运行中的 Instance Store 实例，检测并修复配置漂移"""
    paginator = ec2.get_paginator('describe_instances')
    issues = []

    for page in paginator.paginate(
        Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
    ):
        for res in page['Reservations']:
            for inst in res['Instances']:
                iid = inst['InstanceId']
                itype = inst['InstanceType']

                if not has_instance_store(itype):
                    continue

                # 检查 Stop Protection
                attr = ec2.describe_instance_attribute(
                    InstanceId=iid, Attribute='disableApiStop'
                )
                if not attr['DisableApiStop']['Value']:
                    issues.append(f"{iid} ({itype}): Stop Protection OFF")
                    enable_protection(iid)

                # 检查 Termination Protection
                attr = ec2.describe_instance_attribute(
                    InstanceId=iid, Attribute='disableApiTermination'
                )
                if not attr['DisableApiTermination']['Value']:
                    issues.append(f"{iid} ({itype}): Termination Protection OFF")
                    enable_protection(iid)

    return issues


def handler(event, context):
    """Lambda Handler — 支持实时事件和定时巡检两种触发"""

    # 实时保护：新实例启动
    if event.get('source') == 'aws.ec2':
        detail = event['detail']
        instance_id = detail['instance-id']

        if detail['state'] == 'running':
            resp = ec2.describe_instances(InstanceIds=[instance_id])
            instance = resp['Reservations'][0]['Instances'][0]
            itype = instance['InstanceType']

            if has_instance_store(itype):
                result = enable_protection(instance_id)
                print(f"[AUTO-PROTECT] {instance_id} ({itype}): {result}")

                if SNS_TOPIC_ARN:
                    sns.publish(
                        TopicArn=SNS_TOPIC_ARN,
                        Subject=f'[EC2 Auto-Protect] Protected {instance_id}',
                        Message=f'Instance {instance_id} ({itype}) auto-protected.\n{result}'
                    )

    # 定时巡检
    else:
        issues = audit_all_instances()
        if issues:
            msg = "Instance Store Protection Audit - Issues Found:\n\n" + "\n".join(issues)
            print(f"[AUDIT] {len(issues)} issues found and auto-fixed")
            if SNS_TOPIC_ARN:
                sns.publish(
                    TopicArn=SNS_TOPIC_ARN,
                    Subject=f'[EC2 Auto-Protect Audit] {len(issues)} issues fixed',
                    Message=msg
                )
        else:
            print("[AUDIT] All Instance Store instances properly protected")

    return {'statusCode': 200}

{
    "source": ["aws.ec2"],
    "detail-type": ["EC2 Instance State-change Notification"],
    "detail": {
        "state": ["running"]
    }
}

cron(0 2 * * ? *)    # 每天 UTC 02:00 执行