Event-Driven Automation: Building a Serverless Maintenance Bot with Lambda & EventBridge

Something happens (event)
       │
       ▼
Run a script (function)
       │
       ▼
Done — wait for next event

┌─────────────────────────────────────────────────────┐
│              Serverless Maintenance Bot              │
├─────────────────────────────────────────────────────┤
│                                                     │
│  EC2 State Change ──► auto-snapshot-lambda          │
│  (EventBridge)         (backup EBS on failure)      │
│                                                     │
│  EventBridge Schedule ──► nightly-maintenance-lambda│
│  (cron: 2 AM daily)       (cleanup, reset, report)  │
│                                                     │
│  CloudWatch Alarm ──► auto-remediation-lambda       │
│  (CPU > 80%)             (scale out / restart)      │
│                                                     │
└─────────────────────────────────────────────────────┘

{
  "version": "0",
  "id": "6a7e8feb-b491-4cf7-a9f1-bf3703467718",
  "source": "aws.ec2",
  "detail-type": "EC2 Instance State-change Notification",
  "region": "us-east-1",
  "detail": {
    "instance-id": "i-0123456789abcdef0",
    "state": "stopped"
  }
}

# auto_snapshot.py
import boto3
import json
import logging
from datetime import datetime

logger = logging.getLogger()
logger.setLevel(logging.INFO)

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

ALERT_TOPIC_ARN = os.environ['ALERT_TOPIC_ARN']
SNAPSHOT_TAG_KEY = 'AutoSnapshot'
SNAPSHOT_TAG_VALUE = 'true'


def handler(event, context):
    """
    Triggered by EventBridge on EC2 state change.
    Creates EBS snapshots for all volumes attached to the affected instance.
    """
    detail = event['detail']
    instance_id = detail['instance-id']
    state = detail['state']
    region = event['region']

    logger.info(f'EC2 state change: {instance_id} → {state}')

    # Only act on unexpected stops/reboots (not intentional shutdowns)
    trigger_states = {'stopped', 'stopping'}
    if state not in trigger_states:
        logger.info(f'State {state} does not require snapshot. Skipping.')
        return {'action': 'skipped', 'reason': f'state={state}'}

    # Check if this instance is tagged for auto-snapshot
    instance = ec2.describe_instances(InstanceIds=[instance_id])
    tags = instance['Reservations'][0]['Instances'][0].get('Tags', [])
    tag_map = {t['Key']: t['Value'] for t in tags}

    if tag_map.get(SNAPSHOT_TAG_KEY) != SNAPSHOT_TAG_VALUE:
        logger.info(f'Instance {instance_id} not tagged for auto-snapshot. Skipping.')
        return {'action': 'skipped', 'reason': 'not tagged'}

    instance_name = tag_map.get('Name', instance_id)

    # Get all EBS volumes attached to this instance
    volumes_response = ec2.describe_volumes(
        Filters=[{
            'Name': 'attachment.instance-id',
            'Values': [instance_id]
        }]
    )

    volumes = volumes_response['Volumes']
    snapshot_ids = []
    timestamp = datetime.utcnow().strftime('%Y%m%d-%H%M%S')

    for volume in volumes:
        volume_id = volume['VolumeId']
        device = volume['Attachments'][0]['Device']

        logger.info(f'Creating snapshot for volume {volume_id} ({device})')

        snapshot = ec2.create_snapshot(
            VolumeId=volume_id,
            Description=f'Auto-snapshot: {instance_name} ({instance_id}) state={state}',
            TagSpecifications=[{
                'ResourceType': 'snapshot',
                'Tags': [
                    {'Key': 'Name', 'Value': f'auto-{instance_name}-{device}-{timestamp}'},
                    {'Key': 'InstanceId', 'Value': instance_id},
                    {'Key': 'TriggerState', 'Value': state},
                    {'Key': 'AutoCreated', 'Value': 'true'},
                    {'Key': 'CreatedAt', 'Value': timestamp}
                ]
            }]
        )

        snapshot_ids.append(snapshot['SnapshotId'])
        logger.info(f'Snapshot created: {snapshot["SnapshotId"]} for volume {volume_id}')

    # Send alert notification
    sns.publish(
        TopicArn=ALERT_TOPIC_ARN,
        Subject=f'[Auto-Snapshot] {instance_name} ({state})',
        Message=json.dumps({
            'instance_id': instance_id,
            'instance_name': instance_name,
            'state': state,
            'snapshots_created': snapshot_ids,
            'timestamp': timestamp
        }, indent=2)
    )

    logger.info(f'Auto-snapshot complete: {len(snapshot_ids)} snapshots created')
    return {
        'instance_id': instance_id,
        'state': state,
        'snapshots_created': snapshot_ids
    }

# serverless.yml
functions:
  autoSnapshot:
    handler: auto_snapshot.handler
    environment:
      ALERT_TOPIC_ARN: !Ref OpsAlertTopic
    events:
      - eventBridge:
          pattern:
            source:
              - aws.ec2
            detail-type:
              - EC2 Instance State-change Notification
            detail:
              state:
                - stopped
                - stopping
    iamRoleStatements:
      - Effect: Allow
        Action:
          - ec2:DescribeInstances
          - ec2:DescribeVolumes
          - ec2:CreateSnapshot
          - ec2:CreateTags
          - sns:Publish
        Resource: '*'

# nightly_maintenance.py
import boto3
import json
import logging
import os
from datetime import datetime, timedelta

logger = logging.getLogger()
logger.setLevel(logging.INFO)

ec2 = boto3.client('ec2')
dynamodb = boto3.resource('dynamodb')
s3 = boto3.client('s3')
sns = boto3.client('sns')

RETENTION_DAYS = int(os.environ.get('SNAPSHOT_RETENTION_DAYS', '30'))
METRICS_TABLE = os.environ['METRICS_TABLE']
REPORT_BUCKET = os.environ['REPORT_BUCKET']
ALERT_TOPIC_ARN = os.environ['ALERT_TOPIC_ARN']


def handler(event, context):
    """
    Runs nightly at 2 AM UTC via EventBridge scheduled rule.
    Performs: snapshot cleanup, metrics archival, daily report generation.
    """
    logger.info('Nightly maintenance bot started')
    results = {}

    # Task 1: Clean up old auto-created snapshots
    results['snapshot_cleanup'] = cleanup_old_snapshots()

    # Task 2: Archive daily metrics to S3
    results['metrics_archival'] = archive_daily_metrics()

    # Task 3: Generate and send daily ops report
    results['daily_report'] = send_daily_report(results)

    logger.info(f'Nightly maintenance complete: {json.dumps(results)}')
    return results


def cleanup_old_snapshots() -> dict:
    """Delete auto-created snapshots older than RETENTION_DAYS"""
    cutoff_date = datetime.utcnow() - timedelta(days=RETENTION_DAYS)

    # Find all auto-created snapshots
    response = ec2.describe_snapshots(
        OwnerIds=['self'],
        Filters=[{'Name': 'tag:AutoCreated', 'Values': ['true']}]
    )

    deleted = []
    kept = []

    for snapshot in response['Snapshots']:
        start_time = snapshot['StartTime'].replace(tzinfo=None)

        if start_time < cutoff_date:
            ec2.delete_snapshot(SnapshotId=snapshot['SnapshotId'])
            deleted.append(snapshot['SnapshotId'])
            logger.info(f'Deleted old snapshot: {snapshot["SnapshotId"]} '
                       f'(created: {start_time.date()})')
        else:
            kept.append(snapshot['SnapshotId'])

    logger.info(f'Snapshot cleanup: deleted={len(deleted)}, kept={len(kept)}')
    return {'deleted': len(deleted), 'kept': len(kept)}


def archive_daily_metrics() -> dict:
    """Archive yesterday's metrics from DynamoDB to S3"""
    table = dynamodb.Table(METRICS_TABLE)
    yesterday = (datetime.utcnow() - timedelta(days=1)).strftime('%Y-%m-%d')

    # Scan yesterday's records (use Query with GSI in production)
    response = table.scan(
        FilterExpression='begins_with(#ts, :date)',
        ExpressionAttributeNames={'#ts': 'timestamp'},
        ExpressionAttributeValues={':date': yesterday}
    )

    records = response['Items']

    if records:
        # Write to S3 as JSON Lines
        s3_key = f'metrics-archive/{yesterday}/metrics.jsonl'
        body = '\n'.join(json.dumps(r) for r in records)

        s3.put_object(
            Bucket=REPORT_BUCKET,
            Key=s3_key,
            Body=body.encode('utf-8'),
            ContentType='application/x-ndjson'
        )

        logger.info(f'Archived {len(records)} records to s3://{REPORT_BUCKET}/{s3_key}')

    return {'records_archived': len(records), 'date': yesterday}


def send_daily_report(results: dict) -> dict:
    """Send daily ops summary via SNS"""
    today = datetime.utcnow().strftime('%Y-%m-%d')

    report = {
        'date': today,
        'maintenance_results': results,
        'generated_at': datetime.utcnow().isoformat()
    }

    sns.publish(
        TopicArn=ALERT_TOPIC_ARN,
        Subject=f'[Daily Ops Report] {today}',
        Message=json.dumps(report, indent=2, default=str)
    )

    return {'report_sent': True, 'date': today}

# serverless.yml
functions:
  nightlyMaintenance:
    handler: nightly_maintenance.handler
    timeout: 300          # 5 minutes — cleanup may take time
    environment:
      SNAPSHOT_RETENTION_DAYS: '30'
      METRICS_TABLE: !Ref MetricsTable
      REPORT_BUCKET: !Ref ReportBucket
      ALERT_TOPIC_ARN: !Ref OpsAlertTopic
    events:
      - schedule:
          rate: cron(0 2 * * ? *)    # 2:00 AM UTC every day
          enabled: true
    destinations:
      onFailure: !Ref OpsAlertTopic  # alert if the maintenance bot itself fails

# auto_remediation.py
import boto3
import json
import logging
import os

logger = logging.getLogger()
logger.setLevel(logging.INFO)

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

ALERT_TOPIC_ARN = os.environ['ALERT_TOPIC_ARN']
ASG_NAME = os.environ.get('AUTO_SCALING_GROUP_NAME')


def handler(event, context):
    """
    Triggered by CloudWatch Alarm via SNS → Lambda.
    Parses alarm state and takes automated remediation action.
    """
    # CloudWatch alarms arrive via SNS — parse the message
    for record in event['Records']:
        message = json.loads(record['Sns']['Message'])

        alarm_name = message['AlarmName']
        alarm_state = message['NewStateValue']
        old_state = message['OldStateValue']
        reason = message['NewStateReason']

        logger.info(f'Alarm: {alarm_name} | {old_state} → {alarm_state}')
        logger.info(f'Reason: {reason}')

        if alarm_state != 'ALARM':
            logger.info('Alarm resolved or insufficient data. No action needed.')
            continue

        # Route to appropriate remediation based on alarm name
        if 'high-cpu' in alarm_name.lower():
            result = remediate_high_cpu(alarm_name)
        elif 'disk-space' in alarm_name.lower():
            result = remediate_disk_space(alarm_name)
        else:
            result = {'action': 'no_handler', 'alarm': alarm_name}
            logger.warning(f'No remediation handler for alarm: {alarm_name}')

        # Notify ops team of automated action taken
        sns.publish(
            TopicArn=ALERT_TOPIC_ARN,
            Subject=f'[Auto-Remediation] {alarm_name}',
            Message=json.dumps({
                'alarm': alarm_name,
                'state': alarm_state,
                'reason': reason,
                'automated_action': result
            }, indent=2)
        )

    return {'processed': len(event['Records'])}


def remediate_high_cpu(alarm_name: str) -> dict:
    """Scale out the Auto Scaling Group to handle high CPU load"""
    if not ASG_NAME:
        return {'action': 'skipped', 'reason': 'ASG_NAME not configured'}

    # Get current desired capacity
    asg = autoscaling.describe_auto_scaling_groups(
        AutoScalingGroupNames=[ASG_NAME]
    )['AutoScalingGroups'][0]

    current_desired = asg['DesiredCapacity']
    max_size = asg['MaxSize']
    new_desired = min(current_desired + 2, max_size)  # add 2 instances, respect max

    if new_desired == current_desired:
        logger.warning(f'ASG {ASG_NAME} already at max capacity ({max_size})')
        return {'action': 'at_max_capacity', 'current': current_desired}

    autoscaling.set_desired_capacity(
        AutoScalingGroupName=ASG_NAME,
        DesiredCapacity=new_desired,
        HonorCooldown=False  # bypass cooldown for alarm-triggered scaling
    )

    logger.info(f'Scaled out {ASG_NAME}: {current_desired} → {new_desired}')
    return {
        'action': 'scale_out',
        'asg': ASG_NAME,
        'previous_desired': current_desired,
        'new_desired': new_desired
    }


def remediate_disk_space(alarm_name: str) -> dict:
    """Log disk space alert — automated cleanup is risky, notify instead"""
    logger.warning(f'Disk space alarm: {alarm_name}. Manual review recommended.')
    return {
        'action': 'notification_sent',
        'reason': 'disk cleanup requires manual review'
    }

# serverless.yml
functions:
  autoRemediation:
    handler: auto_remediation.handler
    environment:
      ALERT_TOPIC_ARN: !Ref OpsAlertTopic
      AUTO_SCALING_GROUP_NAME: !Ref AppAutoScalingGroup
    events:
      - sns:
          arn: !Ref CloudWatchAlarmTopic
          topicName: cloudwatch-alarms

resources:
  Resources:
    # CloudWatch alarm that triggers remediation
    HighCpuAlarm:
      Type: AWS::CloudWatch::Alarm
      Properties:
        AlarmName: app-server-high-cpu
        MetricName: CPUUtilization
        Namespace: AWS/EC2
        Statistic: Average
        Period: 300           # 5-minute average
        EvaluationPeriods: 2  # must be high for 10 minutes
        Threshold: 80
        ComparisonOperator: GreaterThanThreshold
        AlarmActions:
          - !Ref CloudWatchAlarmTopic   # → SNS → Lambda
        Dimensions:
          - Name: AutoScalingGroupName
            Value: !Ref AppAutoScalingGroup

# serverless.yml — complete maintenance bot stack
service: serverless-maintenance-bot

provider:
  name: aws
  runtime: python3.12
  region: us-east-1
  iam:
    role:
      statements:
        - Effect: Allow
          Action:
            - ec2:Describe*
            - ec2:CreateSnapshot
            - ec2:DeleteSnapshot
            - ec2:CreateTags
            - autoscaling:Describe*
            - autoscaling:SetDesiredCapacity
            - dynamodb:Scan
            - dynamodb:Query
            - s3:PutObject
            - sns:Publish
          Resource: '*'

functions:
  autoSnapshot:
    handler: auto_snapshot.handler
    environment:
      ALERT_TOPIC_ARN: !Ref OpsAlertTopic
    events:
      - eventBridge:
          pattern:
            source: [aws.ec2]
            detail-type: [EC2 Instance State-change Notification]
            detail:
              state: [stopped, stopping]

  nightlyMaintenance:
    handler: nightly_maintenance.handler
    timeout: 300
    environment:
      SNAPSHOT_RETENTION_DAYS: '30'
      METRICS_TABLE: !Ref MetricsTable
      REPORT_BUCKET: !Ref ReportBucket
      ALERT_TOPIC_ARN: !Ref OpsAlertTopic
    events:
      - schedule:
          rate: cron(0 2 * * ? *)
    destinations:
      onFailure: !Ref OpsAlertTopic

  autoRemediation:
    handler: auto_remediation.handler
    environment:
      ALERT_TOPIC_ARN: !Ref OpsAlertTopic
      AUTO_SCALING_GROUP_NAME: !Ref AppAutoScalingGroup
    events:
      - sns:
          arn: !Ref CloudWatchAlarmTopic
          topicName: cloudwatch-alarms

resources:
  Resources:
    OpsAlertTopic:
      Type: AWS::SNS::Topic
      Properties:
        TopicName: ops-alerts
        Subscription:
          - Protocol: email
            Endpoint: ${env:OPS_EMAIL}

    CloudWatchAlarmTopic:
      Type: AWS::SNS::Topic
      Properties:
        TopicName: cloudwatch-alarms

    MetricsTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: app-metrics
        BillingMode: PAY_PER_REQUEST
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH

    ReportBucket:
      Type: AWS::S3::Bucket
      Properties:
        BucketName: ${self:service}-reports-${aws:accountId}
        LifecycleConfiguration:
          Rules:
            - Id: archive-old-reports
              Status: Enabled
              Transitions:
                - TransitionInDays: 90
                  StorageClass: GLACIER

destinations:
  onFailure: !Ref OpsAlertTopic  # you WILL know when the bot breaks

# Idempotent snapshot creation — check before creating
def create_snapshot_if_not_exists(volume_id: str, instance_id: str) -> str:
    existing = ec2.describe_snapshots(
        Filters=[
            {'Name': 'volume-id', 'Values': [volume_id]},
            {'Name': 'tag:InstanceId', 'Values': [instance_id]},
            {'Name': 'tag:AutoCreated', 'Values': ['true']},
            # Only check snapshots from the last hour
        ],
        OwnerIds=['self']
    )

    if existing['Snapshots']:
        snap_id = existing['Snapshots'][0]['SnapshotId']
        logger.info(f'Snapshot already exists: {snap_id}. Skipping duplicate.')
        return snap_id

    # Create new snapshot
    snapshot = ec2.create_snapshot(VolumeId=volume_id, ...)
    return snapshot['SnapshotId']

functions:
  autoSnapshot:
    handler: auto_snapshot.handler
    deadLetter:
      targetArn: !GetAtt AutoSnapshotDLQ.Arn

Tags=[
    {'Key': 'AutoCreated', 'Value': 'true'},
    {'Key': 'CreatedBy', 'Value': 'serverless-maintenance-bot'},
    {'Key': 'LambdaFunction', 'Value': context.function_name},
    {'Key': 'CreatedAt', 'Value': datetime.utcnow().isoformat()}
]