DEV Community: Prashant Gupta

Automated EC2 Instance Scheduling: A Cost-Optimized Approach to Managing Non-Production Workloads

Prashant Gupta — Fri, 02 Jan 2026 08:48:39 +0000

Automated EC2 Instance Scheduling: A Cost-Optimized Approach to Managing Non-Production Workloads

Introduction & Problem Statement

In cloud environments, cost optimization is a critical concern for engineering teams. One of the most straightforward yet impactful strategies is scheduling non-production EC2 instances to run only during business hours. Consider a typical scenario:

Development/Any environments running 24/7 unnecessarily
Monthly cost: ~$730 per instance (assuming $1/hour)
Actual usage: Only 12 hours/day, 5 days/week (~60 hours/week)
Potential savings: Up to 64% reduction in compute costs

However, manually starting and stopping instances is error-prone, time-consuming, and doesn't scale. Teams need an automated, reliable, and auditable solution that:

Schedules instances based on business hours
Supports multiple modules (backend, frontend, databases)
Provides notifications for visibility and troubleshooting
Integrates with existing infrastructure (VPC, security groups, IAM)
Maintains high availability during operational hours

This article presents a production-ready solution using AWS Lambda, EventBridge, Terraform, and SNS to automate EC2 instance lifecycle management with enterprise-grade reliability.

Architecture & Design Overview

High-Level Architecture

┌─────────────────────────────────────────────────────────────────┐
│                     EventBridge Scheduler                        │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ Start Event  │  │ Stop Event   │  │ Start Event  │          │
│  │ (06:00 UTC)  │  │ (18:00 UTC)  │  │ (06:00 UTC)  │  ...     │
│  │ Backend      │  │ Backend      │  │ Frontend     │          │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘          │
└─────────┼──────────────────┼──────────────────┼─────────────────┘
          │                  │                  │
          └──────────────────┼──────────────────┘
                             ▼
                  ┌──────────────────────┐
                  │   Lambda Function    │
                  │  (VPC-enabled)       │
                  │  - Validate Event    │
                  │  - Start/Stop EC2    │
                  │  - Send Notification │
                  └──────────┬───────────┘
                             │
                ┌────────────┼────────────┐
                ▼            ▼            ▼
         ┌──────────┐  ┌─────────┐  ┌─────────┐
         │   EC2    │  │   SNS   │  │ CloudW. │
         │ Instances│  │  Topic  │  │  Logs   │
         └──────────┘  └─────────┘  └─────────┘

Key Components

Component	Purpose	Technology
EventBridge Rules	Schedule-based triggers for start/stop operations	AWS EventBridge (Cron)
Lambda Function	Orchestrates EC2 operations and notifications	Python 3.14 Runtime
SNS Topic	Delivers success/failure notifications	AWS SNS
IAM Role	Grants Lambda permissions for EC2, SNS, VPC	AWS IAM
Security Group	Controls Lambda network access	AWS VPC
CloudWatch Logs	Centralized logging and monitoring	AWS CloudWatch
CloudWatch Alarms	Alerts on Lambda errors	AWS CloudWatch Alarms
Terraform	Infrastructure as Code for deployment	Terraform 1.14+

Design Principles

Event-Driven Architecture: EventBridge triggers Lambda based on cron schedules
Idempotency: Safe to retry; starting an already-running instance is a no-op
Fail-Fast Validation: Input validation before AWS API calls
Observability: Comprehensive logging and SNS notifications
Security: VPC-enabled Lambda, least-privilege IAM, no hardcoded credentials
Infrastructure as Code: Fully automated deployment via Terraform

Solution Approach

Workflow

The solution follows a straightforward event-driven workflow:

EventBridge Trigger: Cron expression fires at scheduled time (e.g., 06:00 UTC)
Event Payload: Contains action (start/stop), module name, and instance details
Lambda Invocation: Function receives event and validates parameters
EC2 Operation: Calls start_instances() or stop_instances() API
Notification: Publishes success/failure message to SNS topic
Logging: All operations logged to CloudWatch for audit trail

Event Structure

Each EventBridge rule sends a structured JSON payload:

{
  "action": "start",
  "module": "backendserver",
  "instance_details": {
    "i-11223344556677889": "test-prod-backendserver"
  }
}

This design allows:

Multiple instances per event (batch operations)
Module-based grouping for logical separation
Descriptive naming for better observability

Scheduling Strategy

The solution uses cron expressions for precise scheduling:

Start:  cron(0 6 ? * MON-FRI *)   # 06:00 AM UTC, weekdays only
Stop:   cron(0 18 ? * MON-FRI *)  # 06:00 PM UTC, weekdays only

This ensures:

Cost savings during nights and weekends
Availability during business hours
Flexibility to customize per module

Code Walkthrough

Lambda Function Architecture

The refactored Lambda function follows object-oriented principles with clear separation of concerns:

1. EC2InstanceManager Class

Encapsulates EC2 operations with clean interfaces:

class EC2InstanceManager:
    def start_instances(self, instance_ids: List[str]) -> Dict[str, Any]:
        logger.info(f"Starting instances: {instance_ids}")
        return self.ec2.start_instances(InstanceIds=instance_ids)

    def stop_instances(self, instance_ids: List[str]) -> Dict[str, Any]:
        logger.info(f"Stopping instances: {instance_ids}")
        return self.ec2.stop_instances(InstanceIds=instance_ids)

Benefits:

Single responsibility (EC2 operations only)
Easy to mock for unit testing
Reusable across different contexts

2. NotificationService Class

Handles SNS publishing with intelligent message formatting:

class NotificationService:
    def send_notification(
        self, action: str, module: str, 
        instance_details: Dict[str, str],
        status: str = "INFO",
        error_message: Optional[str] = None
    ) -> None:
        subject = f"{self.subject_prefix} | {status} | {action.upper()} - {module}"
        # Format message based on success/failure
        self.sns.publish(TopicArn=self.topic_arn, Subject=subject, Message=message)

Benefits:

Consistent notification format
Graceful handling of missing SNS configuration
Error notifications for troubleshooting

3. Event Validation

Fail-fast validation prevents unnecessary AWS API calls:

def validate_event(event: Dict[str, Any]) -> tuple[str, str, Dict[str, str]]:
    if action not in ['start', 'stop']:
        raise ValueError(f"Invalid action: '{action}'")
    if not instance_details or not isinstance(instance_details, dict):
        raise ValueError("Missing or invalid 'instance_details'")
    return action, module, instance_details

Benefits:

Clear error messages for debugging
Prevents partial operations
Returns 400 status for client errors

4. Lambda Handler

Orchestrates the entire workflow with comprehensive error handling:

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    try:
        action, module, instance_details = validate_event(event)

        if action == 'start':
            response = instance_manager.start_instances(instance_ids)
        else:
            response = instance_manager.stop_instances(instance_ids)

        notification_service.send_notification(...)
        return {'statusCode': 200, 'body': json.dumps({...})}

    except ValueError as e:
        return {'statusCode': 400, 'body': json.dumps({'error': str(e)})}
    except (ClientError, BotoCoreError) as e:
        notification_service.send_notification(..., status="ERROR")
        return {'statusCode': 500, 'body': json.dumps({'error': str(e)})}

Benefits:

Structured error handling (400 vs 500)
Automatic failure notifications
Detailed logging for troubleshooting

Terraform Infrastructure

The Terraform configuration provisions all required AWS resources:

Key Resources

Lambda Function with VPC configuration
IAM Role with EC2, SNS, and VPC permissions
Security Group for Lambda network access
EventBridge Rules for each start/stop schedule
SNS Topic for notifications
CloudWatch Log Group with retention policy
CloudWatch Alarms for error monitoring

Workspace-Based Configuration

The solution uses Terraform workspaces for environment separation:

workspace_prod_ap-south-1.yml  # Production configuration
workspace.yml                   # Common configuration

This enables:

Multi-environment support (dev, staging, prod)
Region-specific settings
Environment-specific schedules

Configuration & Setup Instructions

Prerequisites

AWS Account with appropriate permissions
Terraform 1.14.3 or higher
AWS CLI configured with credentials
S3 Bucket for Terraform state (e.g., test-infra-terraform)
VPC with private subnets
SNS Topic (optional, created automatically)

Step 1: Clone Repository

git clone <repository-url>
cd devops-automation/aws-start-stop-services

Step 2: Configure Workspace

Edit workspace_prod_ap-south-1.yml with your environment details:

workspace:
  aws:
    account_id: "123456789012"        # Your AWS account ID
    region: "ap-south-1"               # Target region
    vpc:
      id: "vpc-xxxxx"                  # VPC ID
      subnet_ids:
        private: ["subnet-xxx", ...]   # Private subnet IDs

  event:
    start-backendserver:
      schedule_expression: "cron(0 6 ? * MON-FRI *)"
      event_input:
        action: "start"
        module: "backendserver"
        instance_details:
          "i-xxxxx": "instance-name"   # Your instance ID and name

Step 3: Update Backend Configuration

Edit aws.tf to match your S3 bucket:

terraform {
  backend "s3" {
    bucket  = "your-terraform-state-bucket"
    key     = "project/app/lambda/start-stop-services/main.tfstate"
    region  = "ap-south-1"
    encrypt = true
  }
}

Step 4: Deploy Infrastructure

# Initialize Terraform
terraform init

# Select workspace
terraform workspace select prod_ap-south-1 || terraform workspace new prod_ap-south-1

# Review plan
terraform plan

# Apply configuration
terraform apply

Or use the provided script:

chmod +x launch.sh
./launch.sh
# Enter: prod_ap-south-1

Step 5: Verify Deployment

# Check Lambda function
aws lambda get-function --function-name prod-project-start-stop-lambda

# Check EventBridge rules
aws events list-rules --name-prefix prod-project

# Check SNS topic
aws sns list-topics | grep start-stop-lambda

Usage Examples

Manual Lambda Invocation

Test the Lambda function manually:

# Start instances
aws lambda invoke \
  --function-name prod-project-start-stop-lambda \
  --payload '{
    "action": "start",
    "module": "backendserver",
    "instance_details": {
      "i-11223344556677889": "test-prod-backendserver"
    }
  }' \
  response.json

# Check response
cat response.json

Scheduled Execution

EventBridge automatically triggers the Lambda function based on cron schedules:

06:00 AM UTC (Mon-Fri): Start backend and frontend servers
06:00 PM UTC (Mon-Fri): Stop backend and frontend servers

SNS Notifications

Subscribers receive email notifications:

Success Notification:

Subject: MyProject | [LAMBDA Notification] | INFO | START - backendserver

Hi,

Successfully started instances for module 'backendserver'

Instance Details:
{
  "i-11223344556677889": "test-prod-backendserver"
}

Failure Notification:

Subject: MyProject | [LAMBDA Notification] | ERROR | START FAILED - backendserver

Error: An error occurred (InvalidInstanceID.NotFound) when calling the StartInstances operation

Module: backendserver
Action: start
Instances: ['i-11223344556677889']

CloudWatch Logs

View execution logs:

aws logs tail /aws/lambda/prod-project-start-stop-lambda --follow

Best Practices Followed

1. Infrastructure as Code

Terraform for reproducible deployments
Version control for all configuration
Workspace isolation for environments

2. Security

VPC-enabled Lambda for network isolation
Least-privilege IAM with specific resource permissions
No hardcoded credentials (environment variables only)
Encrypted S3 backend for Terraform state

3. Observability

Structured logging with log levels
SNS notifications for all operations
CloudWatch alarms for error detection
30-day log retention for compliance

4. Code Quality

PEP 8 compliance for Python code
Type hints for better IDE support
Docstrings for all functions and classes
Error handling with specific exception types
Modular design with single-responsibility classes

5. Reliability

Event invoke config with 0 retries (idempotent operations)
Input validation before AWS API calls
Graceful error handling with appropriate status codes
Timeout configuration (300 seconds)

6. Cost Optimization

Right-sized Lambda (512 MB memory)
VPC endpoints for S3 access (no NAT gateway charges)
Log retention policy (30 days)
Scheduled execution only during business hours

Security & Performance Considerations

Security

IAM Permissions

The Lambda function uses least-privilege permissions:

{
  "Effect": "Allow",
  "Action": [
    "ec2:StartInstances",
    "ec2:StopInstances",
    "ec2:DescribeInstances"
  ],
  "Resource": "*"
}

Recommendation: Restrict to specific instance IDs using resource-based policies:

{
  "Resource": "arn:aws:ec2:ap-south-1:123456789012:instance/i-*"
}

Network Security

VPC-enabled Lambda prevents public internet access
Security group allows only HTTPS egress
VPC endpoints for S3 and SSM (no internet gateway required)

Secrets Management

SNS topic ARN passed via environment variables
No credentials in code or configuration files
AWS SDK uses IAM role credentials automatically

Performance

Lambda Configuration

Memory: 512 MB (sufficient for boto3 operations)
Timeout: 300 seconds (handles multiple instances)
Runtime: Python 3.14 (latest stable version)

Optimization Strategies

Reuse boto3 clients (initialized outside handler)
Batch operations (multiple instances per invocation)
Async notifications (don't block on SNS publish)
VPC endpoints (reduce latency for AWS API calls)

Scaling Considerations

Concurrent executions: Default limit (1000)
EventBridge rules: No limit on number of schedules
Instance limit: Tested with up to 50 instances per event

Common Pitfalls & Troubleshooting

Issue 1: Lambda Timeout

Symptom: Function times out after 300 seconds

Cause: Too many instances in a single event

Solution: Split into multiple EventBridge rules or increase timeout

lambda:
  timeout: 600  # Increase to 10 minutes

Issue 2: VPC Connectivity

Symptom: Unable to connect to endpoint errors

Cause: Missing VPC endpoints or incorrect security group

Solution: Verify VPC endpoints and security group rules

# Check VPC endpoints
aws ec2 describe-vpc-endpoints --filters "Name=vpc-id,Values=vpc-xxxxx"

# Verify security group allows HTTPS egress
aws ec2 describe-security-groups --group-ids sg-xxxxx

Issue 3: Permission Denied

Symptom: AccessDeniedException or UnauthorizedOperation

Cause: Insufficient IAM permissions

Solution: Verify IAM role has required permissions

# Test IAM permissions
aws iam simulate-principal-policy \
  --policy-source-arn arn:aws:iam::123456789012:role/lambda-role \
  --action-names ec2:StartInstances ec2:StopInstances

Issue 4: Instance Not Found

Symptom: InvalidInstanceID.NotFound error

Cause: Incorrect instance ID in configuration

Solution: Verify instance IDs exist in the target region

# List instances
aws ec2 describe-instances --instance-ids i-xxxxx

Issue 5: SNS Notifications Not Received

Symptom: No email notifications

Cause: SNS subscription not confirmed or topic ARN incorrect

Solution: Confirm SNS subscription and verify topic ARN

# List subscriptions
aws sns list-subscriptions-by-topic --topic-arn arn:aws:sns:...

# Verify environment variable
aws lambda get-function-configuration \
  --function-name prod-project-start-stop-lambda \
  --query 'Environment.Variables.SNS_TOPIC_ARN'

Enhancements & Future Improvements

Short-Term Enhancements

Dynamic Scheduling
- Store schedules in DynamoDB
- Update schedules without redeployment
- Per-instance custom schedules
Cost Reporting
- Calculate actual savings
- Send weekly cost reports
- Compare with 24/7 baseline
Health Checks
- Verify instance state after start/stop
- Retry failed operations
- Alert on persistent failures
Multi-Region Support
- Replicate across regions
- Centralized configuration
- Cross-region reporting

Long-Term Improvements

Auto Scaling Integration
- Coordinate with Auto Scaling groups
- Suspend/resume scaling policies
- Maintain desired capacity
RDS Support
- Start/stop RDS instances
- Snapshot before stop
- Multi-AZ considerations
ECS/EKS Support
- Scale ECS services to 0
- Stop EKS node groups
- Preserve task definitions
Self-Service Portal
- Web UI for schedule management
- Role-based access control
- Audit trail and approval workflow
Machine Learning
- Predict optimal schedules based on usage
- Anomaly detection for unexpected usage
- Automatic schedule adjustments

Code Improvements

Unit Tests
- Mock boto3 clients
- Test error scenarios
- Validate event parsing
Integration Tests
- End-to-end workflow testing
- SNS notification verification
- CloudWatch log validation
CI/CD Pipeline
- Automated testing
- Terraform plan on PR
- Automated deployment

Conclusion

Automated EC2 instance scheduling is a simple yet powerful cost optimization strategy that can reduce non-production compute costs by up to 64%. This solution provides:

✅ Production-ready code with comprehensive error handling

✅ Infrastructure as Code for reproducible deployments

✅ Enterprise-grade security with VPC and IAM best practices

✅ Full observability with logging and notifications

✅ Flexible scheduling with EventBridge cron expressions

✅ Scalable architecture supporting multiple modules and instances

Key Takeaways

Event-driven architecture simplifies scheduling logic
Terraform workspaces enable multi-environment management
SNS notifications provide visibility without manual monitoring
VPC-enabled Lambda maintains security without sacrificing functionality
Modular Python code improves maintainability and testability

Getting Started

Clone the repository
Configure your environment in workspace_prod_ap-south-1.yml
Run ./launch.sh to deploy
Monitor CloudWatch logs and SNS notifications
Iterate on schedules based on actual usage patterns

Cost Impact

For a typical setup with 10 EC2 instances:

Before: 10 instances × 24 hours × 30 days × $1/hour = $7,200/month
After: 10 instances × 12 hours × 22 days × $1/hour = $2,640/month
Savings: $4,560/month (63% reduction)

The Lambda function costs are negligible (~$0.20/month for 1,000 invocations).

Resources

AWS Lambda: https://aws.amazon.com/lambda/
EventBridge: https://aws.amazon.com/eventbridge/
Terraform: https://www.terraform.io/
Boto3 Documentation: https://boto3.amazonaws.com/v1/documentation/api/latest/index.html

Author: Prashant Gupta | Cloud Platform Lead

Repository: devops-automation

License: MIT

Questions or feedback? Open an issue or submit a pull request!

AWS ECS Service Task Recycle

Prashant Gupta — Fri, 02 Jan 2026 08:13:30 +0000

AWS ECS Service Task Recycle

Automated Lambda function for recycling AWS ECS service tasks one by one, maintaining service availability during the process. Unlike ECS force deployment which replaces all tasks in parallel, this solution stops and replaces tasks sequentially with configurable wait times.

Overview

This solution provides controlled task recycling for ECS services by:

Stopping tasks one at a time instead of parallel replacement
Waiting for service stability between each task replacement
Optionally maintaining service state by temporarily increasing capacity
Configurable wait time between task replacements

Features

Sequential Task Recycling: Stops and replaces tasks one by one
Service Stability: Waits for stable state after each task replacement
Capacity Management: Optional temporary capacity increase to maintain availability
Autoscaling Support: Handles services with Application Auto Scaling
Flexible Authentication: Multiple AWS credential methods via AWSSession module
Email Notifications: Optional SMTP notifications on completion
CloudFormation Deployment: Infrastructure as code with automated deployment
Zero Retries: EventInvokeConfig set to 0 retry attempts
Comprehensive Logging: Detailed CloudWatch logs for monitoring

Architecture

Lambda Function (Python 3.13)
├── Event-driven execution
├── AWSSession.py (AWS authentication)
├── Notification.py (Email notifications)
└── input.json (Configuration)

Prerequisites

Python 3.13+
AWS CLI configured
IAM permissions for ECS and Application Auto Scaling
SMTP server (optional, for notifications)

Installation

1. Clone Repository

cd aws-ecs-service-task-recycle

2. Configure Settings

Edit input.json with your configuration:

{
  "awsCredentials": {
    "region_name": "us-east-1"
  },
  "smtpCredentials": {
    "host": "smtp.example.com",
    "port": "587",
    "username": "user@example.com",
    "password": "password",
    "from_email": "noreply@example.com"
  },
  "emailNotification": {
    "email_subject": "ECS Service Task Recycle Completed",
    "subject_prefix": "AWS ECS",
    "to": ["admin@example.com"]
  }
}

3. Deploy CloudFormation Stack

chmod +x cloudformation_deploy.sh lambda_build.sh
./cloudformation_deploy.sh

Usage

Lambda Event Parameters

{
  "cluster_name": "my-ecs-cluster",
  "service_name": "my-service",
  "maintain_service_state": true,
  "wait_time": 30
}

Parameters:

cluster_name (required): ECS cluster name
service_name (required): ECS service name
maintain_service_state (optional, default: true): Temporarily increase capacity by 1
wait_time (optional, default: 30): Seconds to wait between task replacements

Invoke Lambda Function

AWS CLI:

aws lambda invoke \
  --function-name ecs-task-recycle-function \
  --payload '{"cluster_name":"my-cluster","service_name":"my-service","maintain_service_state":true,"wait_time":30}' \
  response.json

AWS Console:

Navigate to Lambda → Functions → ecs-task-recycle-function
Test tab → Create test event
Add event JSON and invoke

How It Works

Process Flow

Get Current State: Retrieve service configuration and running tasks
Increase Capacity (if maintain_service_state=true): Add +1 to desired count
Wait for Stability: Ensure new task is running
Recycle Tasks: For each old task:
- Stop the task
- Wait for replacement task to start
- Wait for service stability
- Sleep for configured wait_time
Restore Capacity: Return to original desired count
Send Notification: Email report (if configured)

Example Scenario

Service with 3 tasks:

Initial State: 3 tasks running
↓
Increase to 4 tasks (maintain availability)
↓
Stop task 1 → Wait stable → Sleep 30s
↓
Stop task 2 → Wait stable → Sleep 30s
↓
Stop task 3 → Wait stable → Sleep 30s
↓
Restore to 3 tasks
↓
Complete

Configuration

AWS Credentials (input.json)

Multiple authentication methods supported:

{
  "awsCredentials": {
    "region_name": "us-east-1",
    "profile_name": "my-profile",
    "role_arn": "arn:aws:iam::123456789012:role/MyRole",
    "access_key": "AKIAIOSFODNN7EXAMPLE",
    "secret_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    "session_token": "token"
  }
}

SMTP Configuration (Optional)

{
  "smtpCredentials": {
    "host": "smtp.gmail.com",
    "port": "587",
    "username": "user@gmail.com",
    "password": "app-password",
    "from_email": "noreply@example.com"
  }
}

IAM Permissions

Required permissions (included in CloudFormation):

{
  "Effect": "Allow",
  "Action": [
    "ecs:DescribeServices",
    "ecs:UpdateService",
    "ecs:ListTasks",
    "ecs:StopTask",
    "ecs:DescribeTasks",
    "application-autoscaling:DescribeScalableTargets",
    "application-autoscaling:RegisterScalableTarget"
  ],
  "Resource": "*"
}

CloudFormation Resources

Lambda Function: Python 3.13 runtime, 900s timeout, 256MB memory
IAM Role: Execution role with ECS and Auto Scaling permissions
EventInvokeConfig: MaximumRetryAttempts set to 0
CloudWatch Logs: 7-day retention

Monitoring

CloudWatch Logs

aws logs tail /aws/lambda/ecs-task-recycle-function --follow

Key Log Messages

Starting task recycle for {cluster}/{service}
Original desired count: X, tasks: Y
Recycling task N/M: {task_arn}
Task N recycled, waiting Xs
Task recycle completed successfully

Troubleshooting

Service Not Stabilizing

Increase waiter MaxAttempts in code (default: 40)
Check ECS service health and task definitions
Verify target group health checks

Timeout Errors

Increase Lambda timeout (default: 900s)
Reduce number of tasks or increase wait_time

Authentication Failures

Verify IAM role permissions
Check AWS credentials in input.json
Ensure Lambda execution role is correct

Best Practices

Test in Non-Production: Always test with non-critical services first
Monitor CloudWatch: Watch logs during first execution
Adjust Wait Time: Tune based on application startup time
Use Maintain State: Enable for production services
Schedule Wisely: Run during low-traffic periods

Comparison with Force Deployment

Feature	Force Deployment	Task Recycle
Task Replacement	Parallel	Sequential
Service Disruption	Higher	Lower
Completion Time	Faster	Slower
Control	Limited	Configurable
Wait Between Tasks	No	Yes

Security Considerations

Lambda execution role follows least privilege
No hardcoded credentials in code
SMTP credentials stored in input.json (use Secrets Manager in production)
CloudWatch logs for audit trail
EventInvokeConfig prevents retry storms

Cost Optimization

Lambda execution time: ~(number_of_tasks × wait_time) seconds
CloudWatch Logs: 7-day retention
No additional AWS service costs
Consider scheduling during off-peak hours

Limitations

Maximum Lambda execution time: 15 minutes
Suitable for services with < 20 tasks (with 30s wait time)
Requires stable service for waiter to succeed
No rollback mechanism on failure

Contributing

Contributions welcome! Please follow the repository structure:

Test changes thoroughly
Update documentation
Follow existing code style
Add error handling

Automating CloudWatch Orphan Alarm Detection: A Production-Ready Solution

Prashant Gupta — Fri, 02 Jan 2026 08:07:20 +0000

Automating CloudWatch Orphan Alarm Detection: A Production-Ready Solution

Introduction & Problem Statement

In large-scale AWS environments, CloudWatch alarms are essential for monitoring infrastructure health and triggering alerts when issues arise. However, as infrastructure evolves—EC2 instances are terminated, RDS databases are deleted, load balancers are removed—the alarms monitoring these resources often remain active, creating "orphan alarms."

These orphan alarms present several challenges:

Cost Inefficiency: Each CloudWatch alarm costs money, and orphaned alarms provide no value
Alert Fatigue: Orphan alarms can trigger false alerts, reducing team responsiveness to real issues
Operational Overhead: Manual identification and cleanup is time-consuming and error-prone
Compliance Risks: Outdated monitoring configurations can fail audit requirements

This solution automates the detection and optional deletion of orphan CloudWatch alarms across multiple AWS services, generating comprehensive Excel reports for review and providing flexible authentication options for multi-account environments.

What This Solution Does

Scans CloudWatch alarms across your AWS account
Identifies alarms monitoring deleted or non-existent resources
Generates detailed Excel reports with orphan alarm information
Optionally deletes orphan alarms automatically
Supports multiple AWS services: EC2, RDS, ECS, Lambda, ALB, Target Groups, and SQS
Provides flexible authentication methods for various deployment scenarios
Integrates seamlessly with CI/CD pipelines (Jenkins example included)

Architecture / Design Overview

The solution follows a modular architecture with clear separation of concerns:

┌─────────────────────────────────────────────────────────────┐
│                     Jenkins Pipeline                         │
│                  (Scheduled Execution)                       │
└──────────────────────┬──────────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────────┐
│                   Python Application                         │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐     │
│  │ AWS Session  │  │ Resource     │  │ Alarm        │     │
│  │ Manager      │──│ Discovery    │──│ Analyzer     │     │
│  └──────────────┘  └──────────────┘  └──────────────┘     │
│         │                  │                  │             │
│         └──────────────────┴──────────────────┘             │
│                            │                                │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐     │
│  │ Report       │  │ Notification │  │ Cleanup      │     │
│  │ Generator    │  │ Service      │  │ Service      │     │
│  └──────────────┘  └──────────────┘  └──────────────┘     │
└─────────────────────────────────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────────┐
│                    AWS Services                              │
│  EC2 │ RDS │ ECS │ Lambda │ ALB │ Target Groups │ SQS      │
└─────────────────────────────────────────────────────────────┘

Component Responsibilities

AWS Session Manager: Handles multiple authentication methods (profiles, roles, access keys, STS tokens)
Resource Discovery: Queries AWS APIs to retrieve current active resources
Alarm Analyzer: Compares CloudWatch alarms against active resources to identify orphans
Report Generator: Creates Excel reports with detailed orphan alarm information
Notification Service: Sends email alerts with attached reports
Cleanup Service: Optionally deletes identified orphan alarms

Solution Approach

Detection Strategy

The solution uses a dimension-matching algorithm to identify orphan alarms:

Resource Enumeration: Query all active resources for each supported service
Alarm Collection: Retrieve all CloudWatch alarms with their dimensions
Dimension Mapping: Match alarm dimensions to resource identifiers
Orphan Identification: Flag alarms whose dimensions don't match any active resource
Report Generation: Compile results into actionable reports

Supported Services

Service	Dimension Keys	Example
EC2 Instances	`InstanceId`	`i-0123456789abcdef0`
RDS Clusters	`DBClusterIdentifier`	`production-aurora-cluster`
RDS Instances	`DBInstanceIdentifier`	`production-mysql-01`
ALB	`LoadBalancer`	`app/prod-alb/1234567890`
Target Groups	`TargetGroup`	`targetgroup/prod-tg/abcdef`
ECS Clusters	`ClusterName`	`production-cluster`
ECS Services	`ClusterName`, `ServiceName`	`prod-cluster`, `api-service`
Lambda Functions	`FunctionName`	`data-processor`
Lambda Versions	`FunctionName`, `Resource`	`processor`, `processor:1`
SQS Queues	`QueueName`	`order-processing-queue`

Code Walkthrough

Core Logic: Orphan Detection Algorithm

The heart of the solution is the dimension-matching algorithm in check_alarm_aws_resources_with_resource_list():

def check_alarm_aws_resources_with_resource_list(
    resource_details: Dict[str, List[str]], 
    namespace: str, 
    alarms: List[Dict], 
    service_data: List[Dict]
) -> List[Dict]:
    """
    Identifies orphan alarms by comparing alarm dimensions with active resources.

    Algorithm:
    1. Filter alarms by namespace
    2. Check if alarm has required dimensions
    3. Verify dimension values match an active resource
    4. Flag as orphan if no match found
    """

AWS Session Management

The AWSSession module provides flexible authentication:

# Priority order:
# 1. AWS Profile (for local development)
# 2. IAM Role (for cross-account access)
# 3. Access Keys + Session Token (for temporary credentials)
# 4. Access Keys (for programmatic access)
# 5. Default credentials (for EC2/Lambda IAM roles)

Resource Discovery Pattern

Each service follows a consistent pagination pattern:

def get_service_details(client, max_results: int) -> List[Dict]:
    """Generic pattern for AWS API pagination"""
    resources = []
    next_token = None

    while True:
        # Fetch page of results
        response = client.describe_resources(
            MaxResults=max_results,
            NextToken=next_token if next_token else None
        )

        # Process results
        resources.extend(process_response(response))

        # Check for more pages
        if 'NextToken' not in response:
            break
        next_token = response['NextToken']

    return resources

Report Generation

The Excel report uses xlsxwriter to create structured, readable output:

One worksheet per service with orphan alarms
Columns: AlarmName, Namespace, and all relevant dimensions
Auto-sized columns for readability
Optional automatic deletion of identified orphans

Configuration & Setup Instructions

Prerequisites

# Python 3.8+
python3 --version

# AWS CLI configured (optional)
aws configure

# Required IAM permissions (see iam_policies.json)

Installation

# Clone repository
git clone <repository-url>
cd devops-automation/aws-cw-orphan-alarms

# Create virtual environment
python3 -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

Configuration Files

1. inputs.yml - Runtime Configuration

# AWS Authentication (choose one method)
region_name: "us-east-1"
profile_name: "my-aws-profile"  # For local development
role_arn: ""                     # For cross-account access
access_key: ""                   # For programmatic access
secret_key: ""
session_token: ""                # For temporary credentials

# Cleanup behavior
delete: false  # Set to true to auto-delete orphan alarms

# Email notifications
Email:
  enabled: true
  details:
    subject_prefix: "Production"
    host: "smtp.office365.com:587"
    username: "alerts@company.com"
    password: "secure-password"
    from: "alerts@company.com"
    to:
      - "devops@company.com"
    cc:
      - "platform@company.com"

2. input.json - Service Configuration

This file defines which services to scan and their dimension mappings. The default configuration covers all supported services. Customize to focus on specific services:

{
    "EC2Instance": {
        "Namespace": ["AWS/EC2", "CWAgent"],
        "Dimension": ["InstanceId"],
        "ExcludeDimension": []
    }
}

IAM Permissions

Create an IAM policy with the following permissions:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "cloudwatch:DescribeAlarms",
                "cloudwatch:DeleteAlarms",
                "ec2:DescribeInstances",
                "rds:DescribeDBInstances",
                "rds:DescribeDBClusters",
                "elasticloadbalancing:DescribeLoadBalancers",
                "elasticloadbalancing:DescribeTargetGroups",
                "ecs:ListClusters",
                "ecs:ListServices",
                "lambda:ListFunctions",
                "lambda:ListVersionsByFunction",
                "sqs:ListQueues"
            ],
            "Resource": "*"
        }
    ]
}

Usage Examples

Local Execution

# Dry run (report only, no deletion)
python main.py

# Review generated report
open Inventory.xlsx

# Enable deletion in inputs.yml, then run
python main.py

Jenkins Pipeline Integration

The included Jenkinsfile provides automated scheduling:

triggers {
    cron('0 5 * * 1')  // Every Monday at 5 AM
}

Pipeline Flow:

Clean workspace
Pull latest code from Git
Download configuration from S3
Execute Python script in virtual environment
Send SNS notification on failure

Docker Execution

# Build image
docker build -t orphan-alarm-detector .

# Run container
docker run -v $(pwd)/inputs.yml:/app/inputs.yml \
           -v $(pwd)/output:/app/output \
           orphan-alarm-detector

Lambda Deployment

For serverless execution, package the application:

# Create deployment package
pip install -r requirements.txt -t package/
cp *.py package/
cd package && zip -r ../lambda-function.zip . && cd ..

# Deploy via AWS CLI
aws lambda create-function \
    --function-name orphan-alarm-detector \
    --runtime python3.13 \
    --handler main.lambda_handler \
    --zip-file fileb://lambda-function.zip \
    --role arn:aws:iam::ACCOUNT:role/lambda-execution-role

Best Practices Followed

Code Quality

PEP 8 Compliance: Consistent formatting and naming conventions
Type Hints: Enhanced code readability and IDE support
Docstrings: Comprehensive documentation for all functions
Modular Design: Clear separation of concerns across modules
Error Handling: Graceful degradation with detailed logging

AWS Best Practices

Pagination: Handles large result sets without memory issues
Rate Limiting: Respects AWS API throttling limits
Least Privilege: Minimal IAM permissions required
Multi-Region Support: Configurable region targeting
Credential Security: No hardcoded credentials

Operational Excellence

Logging: Structured logging for troubleshooting
Idempotency: Safe to run multiple times
Dry Run Mode: Test before making changes
Audit Trail: Excel reports provide deletion history
Notification: Email alerts for visibility

Security & Performance Considerations

Security

Credential Management:

Never commit credentials to version control
Use AWS Secrets Manager or Parameter Store for sensitive data
Rotate credentials regularly
Use IAM roles when possible (EC2, Lambda, ECS)

Network Security:

Run in private subnets with VPC endpoints for AWS services
Use security groups to restrict outbound access
Enable VPC Flow Logs for audit trails

Data Protection:

Excel reports may contain sensitive resource names
Store reports in encrypted S3 buckets
Implement lifecycle policies for automatic deletion
Use SES with TLS for email transmission

Performance

Optimization Strategies:

Pagination Tuning: Adjust max_results based on account size

   max_results = 100  # Balance between API calls and memory

Parallel Processing: For multi-region scanning, use threading

   from concurrent.futures import ThreadPoolExecutor

   with ThreadPoolExecutor(max_workers=5) as executor:
       futures = [executor.submit(scan_region, region) for region in regions]

Caching: Cache resource lists for multiple alarm checks

   from functools import lru_cache

   @lru_cache(maxsize=128)
   def get_cached_resources(service: str) -> List[Dict]:
       return fetch_resources(service)

Execution Time Estimates:

Account Size	Resources	Alarms	Execution Time
Small	< 100	< 50	30-60 seconds
Medium	100-1000	50-500	2-5 minutes
Large	1000-5000	500-2000	5-15 minutes
Enterprise	> 5000	> 2000	15-30 minutes

Common Pitfalls & Troubleshooting

Issue: "No credentials found"

Cause: AWS credentials not configured properly

Solution:

# Option 1: Configure AWS CLI
aws configure

# Option 2: Set environment variables
export AWS_ACCESS_KEY_ID=your_key
export AWS_SECRET_ACCESS_KEY=your_secret
export AWS_DEFAULT_REGION=us-east-1

# Option 3: Use IAM role (recommended for EC2/Lambda)
# Attach IAM role to compute resource

Issue: "Rate limit exceeded"

Cause: Too many API calls in short period

Solution:

# Add exponential backoff
import time
from botocore.exceptions import ClientError

def api_call_with_retry(func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return func()
        except ClientError as e:
            if e.response['Error']['Code'] == 'Throttling':
                wait_time = 2 ** attempt
                time.sleep(wait_time)
            else:
                raise

Issue: "Permission denied" errors

Cause: Insufficient IAM permissions

Solution:

Review CloudTrail logs for denied API calls
Add missing permissions to IAM policy
Verify IAM role trust relationships for assumed roles

Issue: Alarms not detected as orphans

Cause: Dimension mismatch or namespace filtering

Solution:

Check alarm dimensions in CloudWatch console
Verify input.json configuration matches alarm structure
Review logs for dimension comparison details

Issue: Email notifications not sending

Cause: SMTP configuration or authentication issues

Solution:

# Verify SMTP settings
Email:
  enabled: true
  details:
    host: "smtp.office365.com:587"  # Include port
    username: "full-email@domain.com"  # Use full email
    password: "app-specific-password"  # Not regular password

Enhancements & Future Improvements

Planned Features

Multi-Region Support:

regions = ['us-east-1', 'us-west-2', 'eu-west-1']
for region in regions:
    scan_region(region)
    generate_report(region)

Multi-Account Scanning:

accounts = [
    {'account_id': '111111111111', 'role': 'OrganizationAccountAccessRole'},
    {'account_id': '222222222222', 'role': 'OrganizationAccountAccessRole'}
]
for account in accounts:
    assume_role(account)
    scan_account(account['account_id'])

Slack Integration:

def send_slack_notification(webhook_url: str, orphan_count: int):
    payload = {
        "text": f"Found {orphan_count} orphan alarms",
        "attachments": [{"color": "warning", "text": "Review report"}]
    }
    requests.post(webhook_url, json=payload)

CloudWatch Dashboard:

Metric: Number of orphan alarms detected
Metric: Cost savings from deleted alarms
Metric: Execution duration and success rate

Terraform/CloudFormation Deployment:

Infrastructure as Code for Lambda deployment
EventBridge rule for scheduled execution
S3 bucket for report storage
SNS topic for notifications

Advanced Filtering:

# Exclude alarms by tag
exclude_tags = {'Environment': 'Production', 'Critical': 'true'}

# Exclude alarms by name pattern
exclude_patterns = [r'^prod-.*', r'.*-critical$']

# Age-based filtering (only flag old orphans)
min_age_days = 7

Community Contributions Welcome

Additional AWS service support (CloudFront, API Gateway, DynamoDB)
Azure and GCP equivalents
Web UI for report visualization
Automated remediation workflows
Integration with ServiceNow/Jira for ticketing

Conclusion

Orphan CloudWatch alarms are an often-overlooked source of waste and operational noise in AWS environments. This automated solution provides a production-ready approach to identifying and managing these orphan alarms at scale.

Key Takeaways:

✅ Cost Optimization: Eliminate unnecessary CloudWatch alarm costs
✅ Operational Efficiency: Reduce alert fatigue and manual cleanup efforts
✅ Flexibility: Multiple authentication methods and deployment options
✅ Safety: Dry-run mode and detailed reporting before deletion
✅ Scalability: Handles large environments with thousands of resources
✅ Integration: Works seamlessly with existing CI/CD pipelines

Getting Started:

Clone the repository and install dependencies
Configure inputs.yml with your AWS credentials
Run in dry-run mode to generate initial report
Review orphan alarms and validate detection accuracy
Enable deletion mode for automated cleanup
Schedule regular execution via Jenkins/Lambda

Production Deployment Checklist:

[ ] IAM permissions configured with least privilege
[ ] Credentials stored securely (Secrets Manager/Parameter Store)
[ ] Email notifications tested and working
[ ] Dry-run executed and results validated
[ ] Backup of critical alarms documented
[ ] Monitoring configured for script execution
[ ] Runbook created for troubleshooting
[ ] Stakeholders notified of automation deployment

This solution has been battle-tested in production environments managing thousands of CloudWatch alarms across multiple AWS accounts. By automating orphan alarm detection, teams can focus on meaningful alerts while maintaining clean, cost-effective monitoring infrastructure.

Author: Prashant Gupta | Cloud Platform Lead

Repository: devops-automation

License: MIT

Questions or Issues? Open an issue on GitHub or reach out to the DevOps team.

Building a Multi-Channel AWS EC2 Spot Instance Interruption Alert System

Prashant Gupta — Thu, 01 Jan 2026 10:42:26 +0000

Building a Multi-Channel AWS EC2 Spot Instance Interruption Alert System

A comprehensive guide to implementing real-time notifications for EC2 Spot Instance interruptions with ECS service impact analysis

Introduction & Problem Statement

AWS EC2 Spot Instances offer significant cost savings—up to 90% compared to On-Demand pricing—making them an attractive option for cost-conscious organizations. However, this cost efficiency comes with a trade-off: Spot Instances can be interrupted with only a 2-minute warning when AWS needs the capacity back.

The Challenge

In production environments running containerized workloads on ECS clusters backed by Spot Instances, these interruptions can cause:

Service disruptions without proper preparation
Lost work if applications aren't designed for graceful shutdowns
Cascading failures when multiple instances are interrupted simultaneously
Operational blindness when teams aren't notified of impending interruptions

The Solution

This article presents a comprehensive, production-ready solution that:

Monitors EC2 Spot Instance interruption warnings in real-time
Identifies running ECS services on affected instances
Delivers intelligent notifications across multiple channels (SNS, Google Chat, SMTP)
Provides actionable information for rapid response

Architecture & Design Overview

Our solution leverages AWS's native event-driven architecture to create a robust, serverless notification system:

┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│   EC2 Spot      │    │   EventBridge    │    │   Lambda        │
│   Interruption  │───▶│   Rule           │───▶│   Function      │
│   Warning       │    │                  │    │                 │
└─────────────────┘    └──────────────────┘    └─────────────────┘
                                                         │
                                                         ▼
                       ┌─────────────────────────────────────────────┐
                       │           Notification Channels             │
                       │  ┌─────────┐ ┌─────────────┐ ┌───────────┐ │
                       │  │   SNS   │ │ Google Chat │ │   SMTP    │ │
                       │  │  Topic  │ │   Webhook   │ │   Email   │ │
                       │  └─────────┘ └─────────────┘ └───────────┘ │
                       └─────────────────────────────────────────────┘

Key Components

EventBridge Rule: Captures EC2 Spot Instance Interruption Warning events
Lambda Function: Processes events, discovers ECS services, and orchestrates notifications
Multi-Channel Notifications: Flexible delivery via SNS, Google Chat, and SMTP
ECS Integration: Intelligent service discovery and impact assessment

Solution Approach

Event-Driven Architecture

We chose an event-driven approach because:

Real-time Response: EventBridge delivers events within seconds
Cost Efficiency: Pay only for actual interruption events
Scalability: Handles multiple simultaneous interruptions automatically
Reliability: AWS-managed infrastructure with built-in retry mechanisms

Multi-Channel Notifications

Different teams prefer different communication channels:

SNS: Integration with existing alerting systems (PagerDuty, OpsGenie)
Google Chat: Real-time team collaboration
SMTP Email: Detailed reports with HTML formatting

Service Impact Analysis

Rather than generic alerts, our solution provides:

Service Identification: Which ECS services are running on the interrupted instance
Impact Assessment: Filtering out daemon services that auto-restart
Actionable Information: Instance details for rapid response

Code Walkthrough

Let's examine the key components that make this solution work:

1. Event Processing & Service Discovery

The core logic processes EventBridge events and discovers affected ECS services:

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """Main handler for spot interruption events."""
    instance_id = event.get("detail", {}).get("instance-id", "Unknown")
    config = _load_configuration()

    # Initialize AWS clients
    session = get_aws_session(config['awsCredentials'])
    aws_clients = {
        'ec2': session.client('ec2'),
        'ecs': session.client('ecs'),
        'sns': session.client('sns')
    }

    # Discover services on interrupted instance
    instance_info = _analyze_instance(aws_clients['ec2'], instance_id)
    service_names = _discover_ecs_services(
        aws_clients['ecs'], instance_id, instance_info.get('cluster_name')
    )

    if service_names:
        _send_notifications(aws_clients, config, instance_info, service_names)

    return {"status": "success", "services_affected": len(service_names)}

2. ECS Service Discovery Logic

The critical logic that identifies which services are running on the interrupted instance:

def _discover_ecs_services(ecs_client, instance_id: str, cluster_name: str) -> List[str]:
    """Find ECS services on the interrupted instance."""
    service_names = []

    # Get container instances in cluster
    container_instances = ecs_client.list_container_instances(cluster=cluster_name)

    for arn in container_instances.get('containerInstanceArns', []):
        # Check if this container instance matches our EC2 instance
        details = ecs_client.describe_container_instances(
            cluster=cluster_name, containerInstances=[arn]
        )

        if details['containerInstances'][0]['ec2InstanceId'] == instance_id:
            # Get tasks running on this container instance
            tasks = ecs_client.list_tasks(cluster=cluster_name, containerInstance=arn)

            for task_arn in tasks.get('taskArns', []):
                service_name = _extract_service_from_task(ecs_client, cluster_name, task_arn)
                if service_name:
                    service_names.append(service_name)

    return list(set(service_names))  # Remove duplicates

3. Multi-Channel Notification Logic

The notification orchestration that sends alerts via multiple channels:

def _send_notifications(aws_clients, config, instance_info, service_names) -> List[str]:
    """Send notifications via enabled channels."""
    message = _format_notification_message(instance_info, service_names)
    notifications_sent = []

    # SNS Notification
    if os.environ.get('ENABLE_SNS', 'false').lower() == 'true':
        try:
            aws_clients['sns'].publish(
                TopicArn=os.environ['SNS_TOPIC_ARN'],
                Subject="🚨 EC2 Spot Instance Interruption Alert",
                Message=message
            )
            notifications_sent.append('SNS')
        except Exception as e:
            logger.error(f"SNS notification failed: {e}")

    # Google Chat Notification
    if os.environ.get('ENABLE_CHAT', 'false').lower() == 'true':
        try:
            response = requests.post(
                os.environ['GOOGLE_CHAT_WEBHOOK'],
                json={"text": message}, timeout=10
            )
            response.raise_for_status()
            notifications_sent.append('Google Chat')
        except Exception as e:
            logger.error(f"Google Chat notification failed: {e}")

    # Email Notification
    if os.environ.get('ENABLE_EMAIL', 'false').lower() == 'true':
        try:
            html_content = f"<html><body><pre>{message}</pre></body></html>"
            send_email(config['smtpCredentials'], config['emailNotification'], html_content)
            notifications_sent.append('Email')
        except Exception as e:
            logger.error(f"Email notification failed: {e}")

    return notifications_sent

4. Flexible AWS Authentication

The session management supports multiple authentication methods:

def get_aws_session(credentials: Dict[str, Any]) -> boto3.Session:
    """Create AWS session with flexible authentication."""
    region = credentials.get("region_name", "us-east-1")

    if credentials.get("profile_name"):
        return boto3.Session(profile_name=credentials["profile_name"], region_name=region)
    elif credentials.get("role_arn"):
        return _create_assumed_role_session(credentials["role_arn"], region)
    elif credentials.get("access_key"):
        return boto3.Session(
            aws_access_key_id=credentials["access_key"],
            aws_secret_access_key=credentials["secret_key"],
            region_name=region
        )
    else:
        return boto3.Session(region_name=region)  # Default credential chain

Configuration & Setup Instructions

Prerequisites

Before deploying the solution, ensure you have:

AWS CLI configured with appropriate permissions
Python 3.13+ for local development
IAM permissions for EC2, ECS, SNS, Lambda, and EventBridge
SMTP credentials if using email notifications
Google Chat webhook if using chat notifications

Step 1: Configuration Setup

Create your input.json configuration file:

{
    "awsCredentials": {
        "region_name": "us-east-1",
        "profile_name": "your-aws-profile"
    },
    "smtpCredentials": {
        "host": "smtp.gmail.com",
        "port": "587",
        "username": "alerts@yourcompany.com",
        "password": "your-app-specific-password",
        "from_email": "alerts@yourcompany.com"
    },
    "emailNotification": {
        "email_subject": "🚨 Spot Instance Interruption Alert",
        "subject_prefix": "PRODUCTION",
        "to": ["devops-team@yourcompany.com"],
        "cc": ["platform-team@yourcompany.com"]
    }
}

Step 2: Deployment Scripts

Update the deployment configuration in cloudformation_deploy.sh:

#!/bin/bash
set -e

# Configuration
STACK_NAME="spot-interruption-alerts"
TEMPLATE_FILE="cloudformation-template.yml"

# Notification Settings
SNS_TOPIC_ARN="arn:aws:sns:us-east-1:123456789012:critical-alerts"
GOOGLE_CHAT_WEBHOOK="https://chat.googleapis.com/v1/spaces/AAAA/messages?key=xxx"
ENABLE_SNS="true"
ENABLE_EMAIL="true"
ENABLE_CHAT="true"

echo "🚀 Deploying Spot Interruption Alert System..."

# Build Lambda package
./lambda_build.sh

# Deploy infrastructure
aws cloudformation deploy \
  --template-file $TEMPLATE_FILE \
  --stack-name $STACK_NAME \
  --parameter-overrides \
    SNSTopicArn="$SNS_TOPIC_ARN" \
    GoogleChatWebhook="$GOOGLE_CHAT_WEBHOOK" \
    EnableSNS="$ENABLE_SNS" \
    EnableEmail="$ENABLE_EMAIL" \
    EnableChat="$ENABLE_CHAT" \
  --capabilities CAPABILITY_NAMED_IAM

# Update Lambda code
FUNCTION_NAME=$(aws cloudformation describe-stacks \
  --stack-name $STACK_NAME \
  --query 'Stacks[0].Outputs[?OutputKey==`LambdaFunctionName`].OutputValue' \
  --output text)

aws lambda update-function-code \
  --function-name $FUNCTION_NAME \
  --zip-file fileb://lambda-deployment.zip

echo "✅ Deployment completed successfully!"

Step 3: Build Script

The lambda_build.sh script packages your Lambda function:

#!/bin/bash

echo "📦 Building Lambda deployment package..."

# Clean previous builds
rm -rf package/ lambda-deployment.zip

# Create package directory
mkdir -p package

# Install dependencies
pip install -r requirements.txt -t package/

# Copy source files
cp *.py package/
cp input.json package/

# Create deployment package
cd package && zip -r ../lambda-deployment.zip . && cd ..

# Cleanup
rm -rf package/

echo "✅ Package created: lambda-deployment.zip"

Step 4: Deploy the Solution

# Make scripts executable
chmod +x cloudformation_deploy.sh lambda_build.sh

# Deploy the complete solution
./cloudformation_deploy.sh

Usage Examples

Example 1: SNS Integration with PagerDuty

Configure SNS topic to integrate with your existing alerting system:

# Create SNS topic
aws sns create-topic --name spot-interruption-alerts

# Subscribe PagerDuty endpoint
aws sns subscribe \
  --topic-arn arn:aws:sns:us-east-1:123456789012:spot-interruption-alerts \
  --protocol https \
  --notification-endpoint https://events.pagerduty.com/integration/xxx/enqueue

Example 2: Google Chat Integration

Set up a Google Chat webhook:

Go to Google Chat → Space Settings → Apps & Integrations
Add Webhook
Copy the webhook URL
Update your deployment script with the webhook URL

Example 3: Custom Email Templates

Customize email notifications by modifying the _format_notification_message function:

def _format_notification_message(instance_info, service_names):
    """Custom email template with company branding."""
    return f"""
    <div style="font-family: Arial; max-width: 600px;">
        <h2 style="color: #d32f2f;">🚨 Spot Instance Interruption Alert</h2>

        <div style="background: #f5f5f5; padding: 15px; border-radius: 5px;">
            <h3>Instance Details</h3>
            <ul>
                <li><strong>Instance ID:</strong> {instance_info['instance_id']}</li>
                <li><strong>Type:</strong> {instance_info['instance_type']}</li>
                <li><strong>AZ:</strong> {instance_info['availability_zone']}</li>
            </ul>
        </div>

        <div style="background: #fff3cd; padding: 15px; border-radius: 5px; margin-top: 15px;">
            <h3>Affected Services</h3>
            <ul>
                {''.join(f'<li>{service}</li>' for service in service_names)}
            </ul>
        </div>

        <p><em>Automated alert from AWS Infrastructure Team</em></p>
    </div>
    """

Best Practices Followed

1. Code Quality & Maintainability

PEP 8 Compliance: Consistent code formatting and naming conventions
Type Hints: Enhanced code readability and IDE support
Modular Design: Separated concerns across multiple modules
Comprehensive Logging: Structured logging with contextual information

2. Error Handling & Resilience

Graceful Degradation: Continue processing even if one notification channel fails
Retry Logic: Built-in retry mechanisms for transient failures
Input Validation: Robust validation of configuration and event data
Exception Handling: Comprehensive error handling with detailed logging

3. Security

Least Privilege IAM: Minimal required permissions for Lambda execution role
Credential Management: Secure handling of SMTP and webhook credentials
No Hardcoded Secrets: All sensitive data passed via environment variables or configuration
Input Sanitization: Proper validation of event data and configuration

4. Performance & Cost Optimization

Efficient Resource Usage: Minimal Lambda memory allocation (128MB)
Fast Execution: Optimized code paths for sub-10-second execution
Event-Driven: Pay only for actual interruption events
Connection Reuse: Efficient AWS client initialization

Security & Performance Considerations

Security Measures

IAM Role Permissions

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Effect": "Allow",
               "Action": [
                   "ec2:DescribeInstances",
                   "ecs:ListContainerInstances",
                   "ecs:DescribeContainerInstances",
                   "ecs:ListTasks",
                   "ecs:DescribeTasks",
                   "ecs:DescribeServices",
                   "sns:Publish"
               ],
               "Resource": "*"
           }
       ]
   }

Credential Security
- SMTP passwords stored in configuration files (consider AWS Secrets Manager for production)
- Google Chat webhooks marked as NoEcho in CloudFormation
- No hardcoded credentials in source code
Network Security
- Lambda function can be deployed in VPC for additional isolation
- HTTPS-only communication for all external APIs
- Webhook URL validation to prevent SSRF attacks

Performance Optimizations

Lambda Configuration
- Memory: 128MB (sufficient for most workloads)
- Timeout: 30 seconds (allows for network retries)
- Runtime: Python 3.13 (latest stable version)
Efficient Processing
- Parallel notification sending where possible
- Early exit for instances without services
- Minimal AWS API calls through intelligent filtering
Cost Considerations
- Lambda Invocations: Only triggered by actual interruption events
- Data Transfer: Minimal outbound data for notifications
- CloudWatch Logs: Structured logging to minimize log volume

Common Pitfalls & Troubleshooting

Issue 1: Lambda Timeout

Symptoms: Lambda function times out before completing notifications

Solutions:

# Increase timeout in CloudFormation template
Timeout: 60  # Increase from 30 to 60 seconds

# Add timeout handling in code
import signal

def timeout_handler(signum, frame):
    raise TimeoutError("Lambda function timeout")

signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(25)  # Set alarm for 25 seconds (5 seconds before Lambda timeout)

Issue 2: SMTP Authentication Failures

Symptoms: Email notifications fail with authentication errors

Solutions:

Gmail: Use App-Specific Passwords instead of regular passwords
Office 365: Enable SMTP AUTH for the mailbox
Corporate SMTP: Verify firewall rules allow outbound SMTP traffic

# Enhanced SMTP error handling
def _send_via_smtp(smtp_config, message, recipients):
    try:
        server = smtplib.SMTP(smtp_config["host"], int(smtp_config["port"]))
        server.set_debuglevel(1)  # Enable debug output
        server.starttls()
        server.login(smtp_config["username"], smtp_config["password"])
        # ... rest of the code
    except smtplib.SMTPAuthenticationError as e:
        logger.error(f"SMTP authentication failed: {e}")
        # Consider fallback notification method
    except Exception as e:
        logger.error(f"SMTP error: {e}")

Issue 3: Missing ECS Services

Symptoms: Alerts sent for instances without detected services

Root Causes:

Incorrect ECS cluster name detection
Services not properly tagged
Daemon services incorrectly filtered

Solutions:

# Enhanced cluster detection
def _get_cluster_name_from_instance(ec2_client, instance_id):
    """Multiple methods to detect ECS cluster name."""
    try:
        response = ec2_client.describe_instances(InstanceIds=[instance_id])
        instance = response['Reservations'][0]['Instances'][0]

        # Method 1: Auto Scaling Group tag
        for tag in instance.get('Tags', []):
            if tag['Key'] == 'aws:autoscaling:groupName':
                return tag['Value']

        # Method 2: ECS Cluster tag
        for tag in instance.get('Tags', []):
            if tag['Key'] == 'ecs:cluster-name':
                return tag['Value']

        # Method 3: Parse from instance name
        instance_name = next(
            (tag['Value'] for tag in instance.get('Tags', []) 
             if tag['Key'] == 'Name'), 
            ''
        )
        if 'ecs' in instance_name.lower():
            return instance_name.split('-')[0]  # Assuming naming convention

    except Exception as e:
        logger.error(f"Failed to detect cluster name: {e}")

    return None

Issue 4: Google Chat Webhook Failures

Symptoms: Google Chat notifications fail silently

Solutions:

def _send_chat_notification(message):
    """Enhanced Google Chat notification with retry logic."""
    webhook_url = os.environ.get('GOOGLE_CHAT_WEBHOOK')
    if not webhook_url:
        logger.warning("Google Chat webhook not configured")
        return False

    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = requests.post(
                webhook_url,
                json={"text": message},
                timeout=10,
                headers={'Content-Type': 'application/json'}
            )
            response.raise_for_status()
            logger.info("Google Chat notification sent successfully")
            return True

        except requests.exceptions.RequestException as e:
            logger.warning(f"Google Chat attempt {attempt + 1} failed: {e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff

    logger.error("Google Chat notification failed after all retries")
    return False

Enhancements & Future Improvements

1. Advanced Service Discovery

Enhance ECS service detection with additional metadata:

def _get_enhanced_service_info(ecs_client, cluster_name, service_name):
    """Get detailed service information for better alerting."""
    try:
        service_details = ecs_client.describe_services(
            cluster=cluster_name,
            services=[service_name]
        )

        if service_details and service_details['services']:
            service = service_details['services'][0]
            return {
                'name': service_name,
                'desired_count': service.get('desiredCount', 0),
                'running_count': service.get('runningCount', 0),
                'pending_count': service.get('pendingCount', 0),
                'task_definition': service.get('taskDefinition', '').split('/')[-1],
                'load_balancers': len(service.get('loadBalancers', [])) > 0
            }
    except Exception as e:
        logger.error(f"Failed to get enhanced service info: {e}")

    return {'name': service_name}

2. Intelligent Alert Suppression

Implement smart alerting to reduce noise:

def _should_send_alert(instance_info, service_names):
    """Determine if alert should be sent based on business rules."""

    # Skip alerts during maintenance windows
    if _is_maintenance_window():
        logger.info("Skipping alert during maintenance window")
        return False

    # Skip alerts for non-critical services
    critical_services = os.environ.get('CRITICAL_SERVICES', '').split(',')
    if critical_services and not any(
        service in critical_services for service in service_names
    ):
        logger.info("No critical services affected, skipping alert")
        return False

    # Skip alerts if cluster has sufficient capacity
    if _has_sufficient_capacity(instance_info.get('cluster_name')):
        logger.info("Cluster has sufficient capacity, skipping alert")
        return False

    return True

3. Metrics and Monitoring

Add CloudWatch custom metrics for better observability:

def _publish_custom_metrics(instance_info, service_names, notifications_sent):
    """Publish custom CloudWatch metrics."""
    cloudwatch = boto3.client('cloudwatch')

    try:
        # Metric: Spot interruptions by instance type
        cloudwatch.put_metric_data(
            Namespace='SpotInterruption',
            MetricData=[
                {
                    'MetricName': 'InterruptionCount',
                    'Dimensions': [
                        {
                            'Name': 'InstanceType',
                            'Value': instance_info.get('instance_type', 'Unknown')
                        }
                    ],
                    'Value': 1,
                    'Unit': 'Count'
                }
            ]
        )

        # Metric: Services affected
        cloudwatch.put_metric_data(
            Namespace='SpotInterruption',
            MetricData=[
                {
                    'MetricName': 'ServicesAffected',
                    'Value': len(service_names),
                    'Unit': 'Count'
                }
            ]
        )

    except Exception as e:
        logger.error(f"Failed to publish metrics: {e}")

4. Integration with External Systems

Add support for additional notification channels:

def _send_slack_notification(message):
    """Send notification to Slack via webhook."""
    webhook_url = os.environ.get('SLACK_WEBHOOK_URL')
    if not webhook_url:
        return False

    slack_message = {
        "text": "Spot Instance Interruption Alert",
        "attachments": [
            {
                "color": "danger",
                "text": message,
                "footer": "AWS Lambda Alert System",
                "ts": int(time.time())
            }
        ]
    }

    try:
        response = requests.post(webhook_url, json=slack_message, timeout=10)
        response.raise_for_status()
        return True
    except Exception as e:
        logger.error(f"Slack notification failed: {e}")
        return False

def _create_jira_ticket(instance_info, service_names):
    """Create JIRA ticket for critical service interruptions."""
    if not _is_critical_interruption(service_names):
        return

    jira_config = {
        'url': os.environ.get('JIRA_URL'),
        'username': os.environ.get('JIRA_USERNAME'),
        'token': os.environ.get('JIRA_TOKEN')
    }

    ticket_data = {
        'fields': {
            'project': {'key': 'INFRA'},
            'summary': f"Spot Instance Interruption: {instance_info['instance_id']}",
            'description': f"Services affected: {', '.join(service_names)}",
            'issuetype': {'name': 'Incident'},
            'priority': {'name': 'High'}
        }
    }

    # Implementation details...

5. Configuration Management

Move to AWS Systems Manager Parameter Store or Secrets Manager:

def _load_configuration_from_ssm():
    """Load configuration from AWS Systems Manager Parameter Store."""
    ssm = boto3.client('ssm')

    try:
        # Get SMTP configuration
        smtp_params = ssm.get_parameters_by_path(
            Path='/spot-interruption/smtp',
            Recursive=True,
            WithDecryption=True
        )

        # Get email configuration
        email_params = ssm.get_parameters_by_path(
            Path='/spot-interruption/email',
            Recursive=True
        )

        # Build configuration dictionary
        config = {
            'smtpCredentials': {},
            'emailNotification': {}
        }

        for param in smtp_params['Parameters']:
            key = param['Name'].split('/')[-1]
            config['smtpCredentials'][key] = param['Value']

        for param in email_params['Parameters']:
            key = param['Name'].split('/')[-1]
            config['emailNotification'][key] = param['Value'].split(',') if key in ['to', 'cc'] else param['Value']

        return config

    except Exception as e:
        logger.error(f"Failed to load configuration from SSM: {e}")
        raise

Conclusion

Building a robust spot instance interruption notification system requires careful consideration of multiple factors: real-time event processing, intelligent service discovery, multi-channel notifications, and operational resilience.

Key Takeaways

Event-Driven Architecture: Leveraging AWS EventBridge provides real-time, scalable event processing with minimal operational overhead.
Multi-Channel Notifications: Different teams prefer different communication methods. Supporting SNS, Google Chat, and SMTP ensures broad coverage.
Intelligent Service Discovery: Rather than generic alerts, providing specific service impact information enables faster response times.
Production-Ready Code: Following Python best practices, comprehensive error handling, and structured logging creates maintainable, reliable systems.
Security First: Implementing least-privilege IAM roles, secure credential handling, and input validation protects against common security vulnerabilities.

Business Impact

This solution provides several business benefits:

Reduced MTTR: Faster notification and service identification reduces mean time to recovery
Cost Optimization: Enables confident use of spot instances with proper monitoring
Operational Excellence: Automated alerting reduces manual monitoring overhead
Scalability: Event-driven architecture scales automatically with infrastructure growth

Next Steps

Consider implementing the suggested enhancements based on your organization's needs:

Advanced Metrics: Add CloudWatch dashboards for spot interruption trends
Integration: Connect with existing ITSM tools (ServiceNow, JIRA)
Machine Learning: Implement predictive analytics for interruption patterns
Multi-Region: Extend solution across multiple AWS regions

The complete source code, CloudFormation templates, and deployment scripts are available in the repository. This solution has been tested in production environments and can be adapted to meet specific organizational requirements.

This solution demonstrates how modern cloud-native architectures can solve real operational challenges while maintaining security, performance, and cost-effectiveness. The modular design allows for easy customization and extension as requirements evolve.

Building a Multi-Channel AWS GuardDuty Alert System: From Detection to Action

Prashant Gupta — Tue, 30 Dec 2025 18:01:51 +0000

Building a Multi-Channel AWS GuardDuty Alert System: From Detection to Action

Introduction & Problem Statement

In today's cloud-first world, security threats are evolving rapidly, and traditional perimeter-based security models are no longer sufficient. AWS GuardDuty provides intelligent threat detection using machine learning, anomaly detection, and integrated threat intelligence, but detecting threats is only half the battle. The real challenge lies in ensuring that security teams are immediately notified when threats are detected, enabling rapid response and mitigation.

The Challenge

Many organizations struggle with:

Alert Fatigue: Security teams receive hundreds of notifications daily
Delayed Response: Critical security findings buried in email or lost in noise
Fragmented Communication: Different teams use different communication channels
Manual Processes: No automated workflow from detection to notification

Our Solution

This article presents a comprehensive, serverless solution that automatically processes AWS GuardDuty findings and delivers intelligent notifications across multiple channels including SNS, email, and Google Chat. The system is designed to be:

Serverless: No infrastructure to manage
Multi-channel: Flexible notification delivery
Configurable: Easy to customize for different environments
Production-ready: Built with enterprise security and reliability in mind

Architecture & Design Overview

Our solution follows a event-driven architecture pattern that leverages AWS native services for maximum reliability and minimal operational overhead.

┌─────────────────┐    ┌──────────────────┐    ┌─────────────────────┐
│   GuardDuty     │    │   EventBridge    │    │   Lambda Function   │
│   Findings      │───▶│      Rule        │───▶│   Processor         │
└─────────────────┘    └──────────────────┘    └─────────────────────┘
                                                          │
                       ┌──────────────────────────────────┼──────────────────────────────────┐
                       │                                  │                                  │
                       ▼                                  ▼                                  ▼
              ┌─────────────────┐              ┌─────────────────┐              ┌─────────────────┐
              │   SNS Topic     │              │   SMTP Server   │              │  Google Chat    │
              │   Notification  │              │   Email Alert   │              │   Webhook       │
              └─────────────────┘              └─────────────────┘              └─────────────────┘

Key Components

EventBridge Rule: Captures GuardDuty findings in real-time
Lambda Function: Processes findings and orchestrates notifications
Multi-Channel Delivery: SNS, Email, and Google Chat integration
Configuration Management: Environment-based settings for flexibility

Solution Approach

Design Principles

Our solution is built on several key principles:

1. Event-Driven Architecture

Leverages AWS EventBridge for real-time event processing
Decoupled components for better maintainability
Automatic scaling based on finding volume

2. Fail-Safe Notifications

Multiple notification channels for redundancy
Graceful degradation if one channel fails
Comprehensive error handling and logging

3. Security First

Least privilege IAM roles
Secure credential management
No hardcoded secrets in code

4. Operational Excellence

Comprehensive logging for troubleshooting
CloudFormation for infrastructure as code
Automated deployment scripts

Code Walkthrough

Let's examine the key components of our solution:

1. AWS Session Management (`AWSSession.py`)

The session manager provides flexible AWS authentication supporting multiple credential types:

import boto3
import logging
from typing import Dict, Any

logger = logging.getLogger(__name__)

def get_aws_session(credentials: Dict[str, Any]) -> boto3.Session:
    """
    Create AWS session with flexible authentication methods.

    Args:
        credentials: Dictionary containing AWS credential information

    Returns:
        boto3.Session: Configured AWS session

    Raises:
        ValueError: If invalid credentials provided
        boto3.exceptions.Boto3Error: If session creation fails
    """
    region = credentials.get("region_name", "us-east-1")

    try:
        if credentials.get("profile_name"):
            logger.info("Creating AWS session with profile authentication")
            return boto3.Session(
                profile_name=credentials["profile_name"],
                region_name=region
            )

        elif credentials.get("role_arn"):
            logger.info("Creating AWS session with assumed role")
            return _create_assumed_role_session(credentials["role_arn"], region)

        elif credentials.get("session_token"):
            logger.info("Creating AWS session with temporary credentials")
            return boto3.Session(
                aws_access_key_id=credentials["access_key"],
                aws_secret_access_key=credentials["secret_key"],
                aws_session_token=credentials["session_token"],
                region_name=region
            )

        elif credentials.get("access_key"):
            logger.info("Creating AWS session with access keys")
            return boto3.Session(
                aws_access_key_id=credentials["access_key"],
                aws_secret_access_key=credentials["secret_key"],
                region_name=region
            )

        else:
            logger.info("Creating AWS session with default credentials")
            return boto3.Session(region_name=region)

    except Exception as e:
        logger.error(f"Failed to create AWS session: {str(e)}")
        raise

def _create_assumed_role_session(role_arn: str, region: str) -> boto3.Session:
    """Create session using assumed role credentials."""
    sts_client = boto3.client('sts', region_name=region)

    response = sts_client.assume_role(
        RoleArn=role_arn,
        RoleSessionName='GuardDutyNotificationSession',
        DurationSeconds=3600
    )

    credentials = response['Credentials']
    return boto3.Session(
        aws_access_key_id=credentials['AccessKeyId'],
        aws_secret_access_key=credentials['SecretAccessKey'],
        aws_session_token=credentials['SessionToken'],
        region_name=region
    )

2. Notification Engine (`Notification.py`)

The notification engine handles email delivery with robust error handling:

import smtplib
import logging
from datetime import datetime
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication
from typing import Dict, List, Any
import os

logger = logging.getLogger(__name__)

def send_email(smtp_config: Dict[str, str], email_details: Dict[str, Any], content: str) -> None:
    """
    Send email notification with comprehensive error handling.

    Args:
        smtp_config: SMTP server configuration
        email_details: Email recipient and subject details
        content: HTML email content

    Raises:
        smtplib.SMTPException: If SMTP operation fails
        ValueError: If invalid email configuration
    """
    logger.info("Initiating email notification")

    try:
        message = _build_email_message(smtp_config, email_details, content)
        recipients = _get_all_recipients(email_details)

        _send_via_smtp(smtp_config, message, recipients)
        logger.info("Email notification sent successfully")

    except Exception as e:
        logger.error(f"Email notification failed: {str(e)}")
        raise

def _build_email_message(smtp_config: Dict[str, str], email_details: Dict[str, Any], content: str) -> MIMEMultipart:
    """Build email message with headers and content."""
    current_date = datetime.now().strftime("%d %B %Y")
    subject = f"{email_details.get('subject_prefix', '')} | {email_details['email_subject']} | {current_date}"

    message = MIMEMultipart()
    message['Subject'] = subject
    message['From'] = smtp_config["from_email"]
    message['To'] = ",".join(email_details.get("to", []))
    message['Cc'] = ",".join(email_details.get("cc", []))

    # Attach HTML content
    message.attach(MIMEText(content, 'html'))

    # Handle attachments if present
    for attachment_path in email_details.get("attachments", []):
        _attach_file(message, attachment_path)

    return message

def _get_all_recipients(email_details: Dict[str, Any]) -> List[str]:
    """Extract all email recipients from configuration."""
    recipients = []
    for field in ['to', 'cc', 'bcc']:
        field_value = email_details.get(field, [])
        if isinstance(field_value, list):
            recipients.extend(field_value)
        elif field_value:
            recipients.append(field_value)

    if not recipients:
        raise ValueError("No valid email recipients configured")

    return recipients

def _send_via_smtp(smtp_config: Dict[str, str], message: MIMEMultipart, recipients: List[str]) -> None:
    """Send email via SMTP server."""
    server = smtplib.SMTP(smtp_config["host"], int(smtp_config["port"]), timeout=30)

    try:
        server.starttls()

        if smtp_config.get("username"):
            server.login(smtp_config["username"], smtp_config["password"])

        server.sendmail(smtp_config["from_email"], recipients, message.as_string())

    finally:
        server.quit()

def _attach_file(message: MIMEMultipart, file_path: str) -> None:
    """Attach file to email message."""
    try:
        with open(file_path, 'rb') as f:
            attachment = MIMEApplication(f.read())

        filename = os.path.basename(file_path)
        attachment.add_header('Content-Disposition', 'attachment', filename=filename)
        message.attach(attachment)

        logger.info(f"File attached: {filename}")

    except Exception as e:
        logger.error(f"Failed to attach file {file_path}: {str(e)}")
        raise

3. Lambda Function Handler (`lambda_function.py`)

The main Lambda function orchestrates the entire notification process:

import json
import os
import logging
from typing import Dict, Any
import requests
from AWSSession import get_aws_session
from Notification import send_email

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """
    Process GuardDuty findings and send multi-channel notifications.

    Args:
        event: EventBridge event containing GuardDuty finding
        context: Lambda context object

    Returns:
        Dict containing response status and details
    """
    logger.info("Processing GuardDuty finding notification")

    try:
        # Load configuration
        config = _load_configuration()

        # Extract finding details
        finding_data = _extract_finding_data(event)

        # Send notifications based on configuration
        notification_results = _send_notifications(config, finding_data)

        logger.info("GuardDuty notification processing completed successfully")

        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Notifications processed successfully',
                'results': notification_results,
                'finding_id': finding_data['finding_id']
            })
        }

    except Exception as e:
        logger.error(f"Failed to process GuardDuty notification: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'error': 'Notification processing failed',
                'message': str(e)
            })
        }

def _load_configuration() -> Dict[str, Any]:
    """Load configuration from input.json file."""
    try:
        with open('input.json', 'r') as f:
            return json.load(f)
    except Exception as e:
        logger.error(f"Failed to load configuration: {str(e)}")
        raise

def _extract_finding_data(event: Dict[str, Any]) -> Dict[str, Any]:
    """Extract relevant data from GuardDuty finding event."""
    detail = event['detail']

    return {
        'severity': detail['severity'],
        'title': detail['title'],
        'description': detail['description'],
        'finding_type': detail['type'],
        'resource': detail['resource']['resourceType'],
        'account': event['account'],
        'region': event['region'],
        'finding_id': detail['id'],
        'created_at': detail.get('createdAt', ''),
        'updated_at': detail.get('updatedAt', '')
    }

def _send_notifications(config: Dict[str, Any], finding_data: Dict[str, Any]) -> Dict[str, str]:
    """Send notifications via configured channels."""
    results = {}

    # SNS Notification
    if _is_enabled('ENABLE_SNS') and os.environ.get('SNS_TOPIC_ARN'):
        try:
            _send_sns_notification(config, finding_data)
            results['sns'] = 'Success'
        except Exception as e:
            logger.error(f"SNS notification failed: {str(e)}")
            results['sns'] = f'Failed: {str(e)}'

    # Email Notification
    if _is_enabled('ENABLE_EMAIL'):
        try:
            _send_email_notification(config, finding_data)
            results['email'] = 'Success'
        except Exception as e:
            logger.error(f"Email notification failed: {str(e)}")
            results['email'] = f'Failed: {str(e)}'

    # Google Chat Notification
    if _is_enabled('ENABLE_CHAT') and os.environ.get('GOOGLE_CHAT_WEBHOOK'):
        try:
            _send_chat_notification(finding_data)
            results['chat'] = 'Success'
        except Exception as e:
            logger.error(f"Chat notification failed: {str(e)}")
            results['chat'] = f'Failed: {str(e)}'

    return results

def _send_sns_notification(config: Dict[str, Any], finding_data: Dict[str, Any]) -> None:
    """Send SNS notification."""
    session = get_aws_session(config['awsCredentials'])
    sns_client = session.client('sns')

    message = _format_sns_message(finding_data)
    subject = f"🚨 GuardDuty Alert | {finding_data['title']}"

    sns_client.publish(
        TopicArn=os.environ['SNS_TOPIC_ARN'],
        Subject=subject,
        Message=message
    )

    logger.info("SNS notification sent successfully")

def _send_email_notification(config: Dict[str, Any], finding_data: Dict[str, Any]) -> None:
    """Send email notification."""
    email_content = _format_email_content(finding_data)

    email_details = config['emailNotification'].copy()
    email_details['email_subject'] = f"GuardDuty Alert | {finding_data['title']}"

    send_email(config['smtpCredentials'], email_details, email_content)
    logger.info("Email notification sent successfully")

def _send_chat_notification(finding_data: Dict[str, Any]) -> None:
    """Send Google Chat notification."""
    message = _format_chat_message(finding_data)

    response = requests.post(
        os.environ['GOOGLE_CHAT_WEBHOOK'],
        json=message,
        timeout=10
    )

    response.raise_for_status()
    logger.info("Google Chat notification sent successfully")

def _format_sns_message(finding_data: Dict[str, Any]) -> str:
    """Format message for SNS notification."""
    return f"""GuardDuty Security Finding

Title: {finding_data['title']}
Description: {finding_data['description']}
Type: {finding_data['finding_type']}
Severity: {finding_data['severity']}
Resource: {finding_data['resource']}
Account: {finding_data['account']}
Region: {finding_data['region']}
Finding ID: {finding_data['finding_id']}

Action Required: Please investigate this finding in the AWS GuardDuty console immediately."""

def _format_email_content(finding_data: Dict[str, Any]) -> str:
    """Format HTML content for email notification."""
    severity_color = _get_severity_color(finding_data['severity'])

    return f"""
    <html>
    <body style="font-family: Arial, sans-serif; margin: 20px;">
        <div style="border-left: 4px solid {severity_color}; padding-left: 20px;">
            <h2 style="color: {severity_color};">🚨 GuardDuty Security Alert</h2>

            <table style="border-collapse: collapse; width: 100%; margin-top: 20px;">
                <tr><td style="padding: 8px; font-weight: bold; background-color: #f5f5f5;">Title:</td>
                    <td style="padding: 8px;">{finding_data['title']}</td></tr>
                <tr><td style="padding: 8px; font-weight: bold; background-color: #f5f5f5;">Description:</td>
                    <td style="padding: 8px;">{finding_data['description']}</td></tr>
                <tr><td style="padding: 8px; font-weight: bold; background-color: #f5f5f5;">Type:</td>
                    <td style="padding: 8px;">{finding_data['finding_type']}</td></tr>
                <tr><td style="padding: 8px; font-weight: bold; background-color: #f5f5f5;">Severity:</td>
                    <td style="padding: 8px; color: {severity_color}; font-weight: bold;">{finding_data['severity']}</td></tr>
                <tr><td style="padding: 8px; font-weight: bold; background-color: #f5f5f5;">Resource:</td>
                    <td style="padding: 8px;">{finding_data['resource']}</td></tr>
                <tr><td style="padding: 8px; font-weight: bold; background-color: #f5f5f5;">Account:</td>
                    <td style="padding: 8px;">{finding_data['account']}</td></tr>
                <tr><td style="padding: 8px; font-weight: bold; background-color: #f5f5f5;">Region:</td>
                    <td style="padding: 8px;">{finding_data['region']}</td></tr>
                <tr><td style="padding: 8px; font-weight: bold; background-color: #f5f5f5;">Finding ID:</td>
                    <td style="padding: 8px; font-family: monospace;">{finding_data['finding_id']}</td></tr>
            </table>

            <div style="margin-top: 20px; padding: 15px; background-color: #fff3cd; border: 1px solid #ffeaa7; border-radius: 4px;">
                <strong>Action Required:</strong> Please investigate this finding in the AWS GuardDuty console immediately.
            </div>
        </div>
    </body>
    </html>
    """

def _format_chat_message(finding_data: Dict[str, Any]) -> Dict[str, str]:
    """Format message for Google Chat notification."""
    return {
        "text": f"""🚨 *GuardDuty Security Alert*

*Title:* {finding_data['title']}
*Description:* {finding_data['description']}
*Type:* {finding_data['finding_type']}
*Severity:* {finding_data['severity']}
*Resource:* {finding_data['resource']}
*Account:* {finding_data['account']}
*Region:* {finding_data['region']}
*Finding ID:* {finding_data['finding_id']}

⚠️ *Action Required:* Please investigate this finding in the AWS GuardDuty console immediately."""
    }

def _get_severity_color(severity: float) -> str:
    """Get color code based on severity level."""
    if severity >= 7.0:
        return "#dc3545"  # Red for high severity
    elif severity >= 4.0:
        return "#fd7e14"  # Orange for medium severity
    else:
        return "#28a745"  # Green for low severity

def _is_enabled(env_var: str) -> bool:
    """Check if notification channel is enabled."""
    return os.environ.get(env_var, 'false').lower() == 'true'

Configuration & Setup Instructions

Prerequisites

AWS CLI configured with appropriate permissions
Python 3.9+ for local development
Access to AWS Lambda, EventBridge, and SNS services

Step 1: Clone and Prepare

git clone <repository-url>
cd aws-guardduty-notification

Step 2: Configure Input Parameters

Update input.json with your environment-specific settings:

{
    "awsCredentials": {
        "region_name": "us-east-1",
        "profile_name": "your-aws-profile"
    },
    "smtpCredentials": {
        "host": "smtp.gmail.com",
        "port": "587",
        "username": "your-email@company.com",
        "password": "your-app-password",
        "from_email": "alerts@company.com"
    },
    "emailNotification": {
        "email_subject": "GuardDuty Security Alert",
        "subject_prefix": "SECURITY",
        "to": ["security-team@company.com"],
        "cc": ["devops@company.com"],
        "bcc": []
    }
}

Step 3: Deploy Infrastructure

# Make deployment script executable
chmod +x cloudformation_deploy.sh

# Update deployment parameters
export SNS_TOPIC_ARN="arn:aws:sns:us-east-1:123456789012:security-alerts"
export GOOGLE_CHAT_WEBHOOK="https://chat.googleapis.com/v1/spaces/xxx/messages?key=xxx"

# Deploy the stack
./cloudformation_deploy.sh

Step 4: Verify Deployment

# Check stack status
aws cloudformation describe-stacks --stack-name guardduty-notifications

# Test Lambda function
aws lambda invoke --function-name guardduty-notifications-guardduty-processor \
  --payload file://test-event.json response.json

Usage Examples

Testing with Sample Event

Create a test event file (test-event.json):

{
    "version": "0",
    "id": "test-event-id",
    "detail-type": "GuardDuty Finding",
    "source": "aws.guardduty",
    "account": "123456789012",
    "time": "2025-01-15T10:30:00Z",
    "region": "us-east-1",
    "detail": {
        "id": "test-finding-id",
        "type": "Backdoor:EC2/C&CActivity.B!DNS",
        "title": "EC2 instance is querying a domain name associated with a known command and control server",
        "description": "EC2 instance i-1234567890abcdef0 is querying a domain name associated with a known command and control server.",
        "severity": 8.5,
        "resource": {
            "resourceType": "Instance"
        },
        "createdAt": "2025-01-15T10:30:00.000Z",
        "updatedAt": "2025-01-15T10:30:00.000Z"
    }
}

Environment Variables Configuration

The Lambda function uses these environment variables:

SNS_TOPIC_ARN=arn:aws:sns:us-east-1:123456789012:security-alerts
GOOGLE_CHAT_WEBHOOK=https://chat.googleapis.com/v1/spaces/xxx/messages?key=xxx
ENABLE_SNS=true
ENABLE_EMAIL=true
ENABLE_CHAT=true

Best Practices Followed

1. Security Best Practices

Least Privilege IAM Roles

Lambda execution role has minimal required permissions
No hardcoded credentials in source code
Secure parameter handling for sensitive data

Credential Management

Support for multiple AWS authentication methods
Secure SMTP credential handling
Environment variable usage for configuration

2. Operational Excellence

Comprehensive Logging

Structured logging throughout the application
Error tracking and debugging information
Performance monitoring capabilities

Infrastructure as Code

Complete CloudFormation template
Automated deployment scripts
Version-controlled infrastructure

3. Reliability & Performance

Error Handling

Graceful degradation when channels fail
Retry logic for transient failures
Comprehensive exception handling

Scalability

Serverless architecture for automatic scaling
Efficient resource utilization
Minimal cold start impact

Security & Performance Considerations

Security Measures

Data Protection
- No sensitive data logged
- Secure credential transmission
- Encrypted communication channels
Access Control
- IAM role-based permissions
- Resource-level access restrictions
- Audit trail maintenance
Network Security
- VPC deployment options
- Security group configurations
- Encrypted data in transit

Performance Optimizations

Lambda Optimization
- Minimal package size
- Efficient memory allocation
- Connection pooling for SMTP
Cost Management
- Pay-per-use pricing model
- Efficient resource utilization
- Automated scaling

Common Pitfalls & Troubleshooting

Issue 1: Email Delivery Failures

Symptoms:

Email notifications not received
SMTP authentication errors

Solutions:

# Check SMTP credentials
aws logs filter-log-events --log-group-name /aws/lambda/guardduty-processor \
  --filter-pattern "SMTP"

# Verify email configuration
python -c "
import smtplib
server = smtplib.SMTP('smtp.gmail.com', 587)
server.starttls()
server.login('username', 'password')
print('SMTP connection successful')
"

Issue 2: Lambda Timeout Errors

Symptoms:

Function timeouts during execution
Incomplete notification delivery

Solutions:

Increase Lambda timeout in CloudFormation template
Optimize SMTP connection handling
Implement asynchronous processing for multiple channels

Issue 3: EventBridge Rule Not Triggering

Symptoms:

GuardDuty findings not triggering Lambda
Missing event pattern matches

Solutions:

# Verify EventBridge rule
aws events describe-rule --name guardduty-notifications-guardduty-rule

# Check rule targets
aws events list-targets-by-rule --rule guardduty-notifications-guardduty-rule

# Test event pattern
aws events test-event-pattern --event-pattern file://event-pattern.json \
  --event file://test-event.json

Issue 4: Google Chat Webhook Failures

Symptoms:

Chat notifications not appearing
Webhook authentication errors

Solutions:

Verify webhook URL format and permissions
Check Google Chat space configuration
Test webhook independently:

curl -X POST "https://chat.googleapis.com/v1/spaces/xxx/messages?key=xxx" \
  -H "Content-Type: application/json" \
  -d '{"text": "Test message"}'

Enhancements & Future Improvements

Planned Enhancements

Advanced Filtering
- Severity-based routing
- Finding type categorization
- Custom filtering rules
Integration Expansions
- Slack integration
- Microsoft Teams support
- PagerDuty escalation
Analytics & Reporting
- Finding trend analysis
- Response time metrics
- Dashboard integration
Automation Workflows
- Automatic remediation for low-risk findings
- Ticket creation integration
- Escalation workflows

Implementation Roadmap

Phase 1: Enhanced Filtering (Q2 2025)

def should_notify(finding_data: Dict[str, Any], filters: Dict[str, Any]) -> bool:
    """Determine if finding should trigger notification based on filters."""
    severity_threshold = filters.get('min_severity', 0)
    excluded_types = filters.get('excluded_types', [])

    if finding_data['severity'] < severity_threshold:
        return False

    if finding_data['finding_type'] in excluded_types:
        return False

    return True

Phase 2: Slack Integration (Q3 2025)

def send_slack_notification(webhook_url: str, finding_data: Dict[str, Any]) -> None:
    """Send formatted notification to Slack channel."""
    payload = {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"🚨 GuardDuty Alert: {finding_data['title']}"
                }
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*Severity:* {finding_data['severity']}"},
                    {"type": "mrkdwn", "text": f"*Account:* {finding_data['account']}"}
                ]
            }
        ]
    }

    requests.post(webhook_url, json=payload)

Phase 3: Automated Remediation (Q4 2025)

Integration with AWS Systems Manager
Automated EC2 instance isolation
Security group modification workflows

Conclusion

Building a robust security notification system requires careful consideration of multiple factors: reliability, security, performance, and operational excellence. Our AWS GuardDuty notification solution demonstrates how to leverage serverless architecture and AWS native services to create a production-ready system that scales automatically and provides comprehensive security alerting.

Key Takeaways

Event-Driven Architecture: Leveraging EventBridge provides real-time processing with minimal operational overhead
Multi-Channel Redundancy: Supporting multiple notification channels ensures critical alerts are never missed
Security First: Implementing proper IAM roles and credential management from the start
Operational Excellence: Comprehensive logging and infrastructure as code enable reliable operations

Getting Started

The complete solution is available in our repository with detailed setup instructions. Start with the basic configuration and gradually enable additional notification channels based on your team's needs.

Community & Support

We encourage contributions and feedback from the community. Whether you're implementing additional notification channels, improving error handling, or adding new features, your contributions help make this solution better for everyone.

Next Steps:

Deploy the basic solution in your environment
Customize notification channels for your team
Implement additional filtering based on your security requirements
Consider contributing enhancements back to the community

Security is a team effort, and automated notification systems like this one help ensure that your security team can respond quickly to threats. By following the patterns and practices outlined in this article, you can build robust, scalable security automation that grows with your organization's needs.

This solution has been tested in production environments and follows AWS Well-Architected Framework principles. For questions or contributions, please refer to the repository's issue tracker and contribution guidelines.

AWS Bedrock AgentCore Runtime: Building and Deploying AI Agents at Scale

Prashant Gupta — Tue, 30 Dec 2025 12:23:03 +0000

AWS Bedrock AgentCore Runtime: Building and Deploying AI Agents at Scale

🚀 Introduction

AWS Bedrock AgentCore Runtime represents a revolutionary approach to deploying and managing AI agents in the cloud. This comprehensive guide walks you through building, deploying, and invoking AI agents using AWS's fully managed AgentCore service, eliminating the complexity of infrastructure management while providing enterprise-grade scalability and security.

In this blog-style guide, we'll explore how to create a production-ready AI agent that leverages Claude 3 Haiku through Bedrock, deploy it using automated build scripts, and invoke it seamlessly from your applications or Lambda functions.

📋 Table of Contents

Architecture Overview
Prerequisites
Project Structure
Core Components
Step-by-Step Implementation
Deployment Process
Agent Invocation
Best Practices
Troubleshooting
Advanced Configuration

🏗️ Architecture Overview

AWS Bedrock AgentCore Runtime provides a serverless, fully managed environment for running AI agents. The architecture consists of several key components:

High-Level Architecture

The AgentCore Runtime follows a simple yet powerful pattern:

Agent Code: Your Python application using the Bedrock AgentCore SDK
S3 Storage: Packaged agent code stored as ZIP files
AgentCore Runtime: Managed service that hosts and executes your agent
Invocation Layer: Direct API calls or Lambda integration for triggering agents

Key Benefits

Serverless: No infrastructure to manage
Scalable: Automatic scaling based on demand
Secure: Built-in AWS security and compliance
Cost-Effective: Pay only for what you use
Integrated: Native integration with Bedrock models

🔧 Prerequisites

Before diving into the implementation, ensure you have the following:

AWS Requirements

AWS Account with appropriate permissions
AWS CLI configured with credentials
Access to Amazon Bedrock service
S3 bucket for storing agent packages

Development Environment

Python 3.13 or higher
pip package manager
bash shell (for deployment scripts)
Git for version control

Required AWS Permissions

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "bedrock:*",
                "bedrock-agentcore:*",
                "s3:GetObject",
                "s3:PutObject",
                "s3:ListBucket"
            ],
            "Resource": "*"
        }
    ]
}

📁 Project Structure

Our AgentCore Runtime project follows a clean, organized structure:

aws-agentcore-runtime/
├── main.py                    # Core agent implementation
├── requirements.txt           # Python dependencies
├── script.sh                 # Automated build and deployment script
├── trigger-agent.py          # Agent invocation client
├── images/                   # Documentation screenshots
│   ├── 1-AgentCoreDiagram.png
│   ├── 2-AgentCoreRuntimeHost.png
│   ├── 3-AgentCoreRuntimeHostSource.png
│   ├── 4-AgentCoreRuntimeHostPermission.png
│   ├── 5-AgentCoreRuntimeHostProtocol.png
│   ├── 6-AgentCoreRuntimeHostSecurity.png
│   └── 7-AgentCoreRuntimeEndpoint.png
└── README.md                 # This comprehensive guide

🧩 Core Components

1. Agent Implementation (`main.py`)

The heart of our AI agent is a Python application that leverages the Bedrock AgentCore SDK:

"""AWS Bedrock AgentCore Runtime - AI Agent Implementation"""

import logging
from typing import Dict, Any

from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent
from strands.models import BedrockModel

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize Bedrock model
bedrock_model = BedrockModel(
    model_id="anthropic.claude-3-haiku-20240307-v1:0",
    region_name="us-east-1",
)

# Initialize app and agent
app = BedrockAgentCoreApp()
agent = Agent(model=bedrock_model)

@app.entrypoint
def invoke(payload: Dict[str, Any]) -> Dict[str, str]:
    """AI agent entrypoint function."""
    try:
        user_message = payload.get("prompt", "Hello! How can I help you today?")
        logger.info(f"Processing request with prompt length: {len(user_message)}")

        result = agent(user_message)

        logger.info("Successfully processed agent request")
        return {"result": result.message}

    except Exception as e:
        logger.error(f"Agent processing failed: {str(e)}")
        return {"result": f"Error processing request: {str(e)}"}

if __name__ == "__main__":
    logger.info("Starting AWS Bedrock AgentCore Runtime")
    app.run()

Key Features:

Error Handling: Comprehensive exception handling with logging
Type Hints: Full type annotations for better code quality
Logging: Structured logging for debugging and monitoring
Flexibility: Configurable model and region settings

2. Dependencies (`requirements.txt`)

Our agent requires specific Python packages:

bedrock-agentcore
strands-agents
pydantic
pydantic_core

Package Overview:

bedrock-agentcore: Core SDK for AgentCore Runtime
strands-agents: Agent framework for building AI applications
pydantic: Data validation and settings management
pydantic_core: Core validation library

3. Build and Deployment Script (`script.sh`)

The automated deployment script handles the entire build and upload process:

#!/bin/bash
set -e

S3_BUCKET=$1
VERSION=$2

if [ -z "$S3_BUCKET" ] || [ -z "$VERSION" ]; then
  echo "Usage: $0 <s3-bucket> <version>"
  exit 1
fi

echo "Building agent version $VERSION for S3 bucket $S3_BUCKET"

# Version normalization
VERSION=$(echo "$VERSION" | tr '[:upper:]' '[:lower:]' | tr ' ' '_' | tr -cd '[:alnum:]-._' | tr '.' '_')

# Clean environment
echo "Cleaning up local files"
rm -f agent.zip
rm -rf package
rm -rf .venv

# Build process
echo "Building agent"
python3 -m venv .venv
source .venv/bin/activate
pip install --upgrade pip --no-cache-dir
pip install -r requirements.txt --no-cache-dir
deactivate

# Package creation
echo "Packaging agent"
mkdir package || true
cp -r .venv/lib/python3.*/site-packages/* package/
cp -r main.py package/
cd package
find . -type f -name '*.py[co]' -delete -o -type d -name __pycache__ -delete
zip -r ../agent.zip *
cd ..

# S3 upload
echo "Uploading agent to S3"
aws s3 cp agent.zip s3://$S3_BUCKET/agent-$VERSION.zip
echo "Agent uploaded to s3://$S3_BUCKET/agent-$VERSION.zip"

# Cleanup
rm -f agent.zip
rm -rf package
rm -rf .venv

echo "Script completed successfully"

Script Features:

Version Management: Automatic version normalization
Clean Builds: Removes previous artifacts
Dependency Isolation: Uses virtual environments
Optimization: Removes unnecessary files from package
Error Handling: Fails fast on any error

🚀 Step-by-Step Implementation

Step 1: Setting Up the AgentCore Runtime

First, create your AgentCore Runtime in the AWS Console:

Navigate to Amazon Bedrock in the AWS Console
Select "AgentCore" from the left navigation
Click "Create Runtime"
Configure the runtime settings:
- Name: Choose a descriptive name
- Description: Document the runtime purpose
- Source: Select "S3" as the source type

Step 2: Configuring Permissions

Set up the necessary IAM permissions:

Create an execution role for the AgentCore Runtime
Attach policies for:
- Bedrock model access
- S3 bucket access
- CloudWatch logging
Configure the runtime to use this role

Step 3: Protocol Configuration

Configure the communication protocol:

Set the invocation method (HTTP/HTTPS)
Configure timeout settings
Set up retry policies
Define payload format requirements

Step 4: Security Configuration

Implement security best practices:

Enable encryption in transit and at rest
Configure VPC settings if required
Set up access controls
Enable audit logging

Step 5: Endpoint Configuration

Finalize the endpoint configuration:

Review the generated endpoint ARN
Configure any custom domains
Set up monitoring and alerting
Test the endpoint connectivity

📦 Deployment Process

Building and Deploying Your Agent

Prepare Your Environment

   # Clone or navigate to your project directory
   cd aws-agentcore-runtime

   # Ensure AWS CLI is configured
   aws configure list

Run the Deployment Script

   # Make the script executable
   chmod +x script.sh

   # Deploy your agent
   ./script.sh your-s3-bucket-name v1.0.0

Monitor the Deployment

   # Check S3 upload
   aws s3 ls s3://your-s3-bucket-name/

   # Verify the agent package
   aws s3 ls s3://your-s3-bucket-name/agent-v1_0_0.zip

Deployment Best Practices

Version Control: Always use semantic versioning
Testing: Test in development environments first
Rollback Plan: Keep previous versions for quick rollback
Monitoring: Set up CloudWatch alarms for deployment health

🔧 Agent Invocation

Direct Invocation from Applications

Use the enhanced trigger-agent.py for invoking your agent:

"""AWS Bedrock AgentCore Runtime - Agent Invocation Client"""

import json
import logging
from typing import Dict, Any

import boto3
from botocore.exceptions import ClientError

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration
REGION_NAME = 'us-east-1'
AGENT_RUNTIME_ARN = 'arn:aws:bedrock-agentcore:us-east-1:999999999999:runtime/hosted_agent_phmxx-dtrZCGFffv'
RUNTIME_SESSION_ID = 'dfmeoagmreaklgmrkleafremoigrmtesogmtrskhmtkrlshmt'  # Must be 33+ chars
QUALIFIER = "DEFAULT"

def invoke_agent(prompt: str) -> Dict[str, Any]:
    """Invoke the Bedrock AgentCore runtime with a given prompt."""
    try:
        client = boto3.client('bedrock-agentcore', region_name=REGION_NAME)
        payload = json.dumps({"prompt": prompt})

        logger.info(f"Invoking agent with prompt length: {len(prompt)}")

        response = client.invoke_agent_runtime(
            agentRuntimeArn=AGENT_RUNTIME_ARN,
            runtimeSessionId=RUNTIME_SESSION_ID,
            payload=payload,
            qualifier=QUALIFIER
        )

        response_body = response['response'].read()
        response_data = json.loads(response_body)

        logger.info("Successfully received agent response")
        return response_data

    except ClientError as e:
        logger.error(f"AWS API error: {e}")
        raise
    except json.JSONDecodeError as e:
        logger.error(f"JSON decode error: {e}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise

def main():
    """Main function to demonstrate agent invocation."""
    try:
        prompt = "Explain machine learning in simple terms"
        response_data = invoke_agent(prompt)

        print("Agent Response:", response_data)

    except Exception as e:
        logger.error(f"Failed to invoke agent: {e}")
        return 1

    return 0

if __name__ == "__main__":
    exit(main())

Lambda Integration

For serverless applications, integrate with AWS Lambda:

import json
import boto3
from typing import Dict, Any

def lambda_handler(event: Dict[str, Any], context) -> Dict[str, Any]:
    """Lambda function to invoke AgentCore Runtime."""

    client = boto3.client('bedrock-agentcore')

    # Extract prompt from event
    prompt = event.get('prompt', 'Hello!')

    # Invoke agent
    response = client.invoke_agent_runtime(
        agentRuntimeArn=event['agentRuntimeArn'],
        runtimeSessionId=event.get('sessionId', 'default-session-id-12345678901234567890'),
        payload=json.dumps({"prompt": prompt}),
        qualifier=event.get('qualifier', 'DEFAULT')
    )

    # Process response
    response_body = response['response'].read()
    response_data = json.loads(response_body)

    return {
        'statusCode': 200,
        'body': json.dumps(response_data)
    }

🎯 Best Practices

Code Quality

Type Hints: Use comprehensive type annotations
Error Handling: Implement robust exception handling
Logging: Add structured logging throughout your code
Documentation: Include docstrings for all functions

Security

Credentials: Never hardcode credentials in your code
IAM Roles: Use least privilege principle for IAM roles
Encryption: Enable encryption for data in transit and at rest
Secrets Management: Use AWS Secrets Manager for sensitive data

Performance

Model Selection: Choose appropriate Bedrock models for your use case
Caching: Implement response caching where appropriate
Timeout Configuration: Set reasonable timeout values
Resource Optimization: Monitor and optimize resource usage

Monitoring and Observability

CloudWatch Metrics: Set up custom metrics for your agent
Alarms: Create alarms for error rates and latency
Distributed Tracing: Use AWS X-Ray for request tracing
Log Analysis: Implement log aggregation and analysis

🔍 Troubleshooting

Common Issues and Solutions

1. Agent Deployment Failures

Problem: Agent fails to deploy to S3 or AgentCore Runtime
Solutions:

Verify S3 bucket permissions
Check AWS CLI configuration
Validate package structure
Review CloudWatch logs

2. Invocation Errors

Problem: Agent invocation returns errors
Solutions:

Verify AgentCore Runtime ARN
Check session ID format (must be 33+ characters)
Validate payload structure
Review IAM permissions

3. Performance Issues

Problem: Slow response times or timeouts
Solutions:

Optimize model selection
Implement request caching
Adjust timeout configurations
Monitor resource utilization

4. Authentication Failures

Problem: Access denied errors
Solutions:

Verify IAM role permissions
Check AWS credentials configuration
Validate resource ARNs
Review security group settings

Debugging Tips

Enable Detailed Logging

   import logging
   logging.basicConfig(level=logging.DEBUG)

Use AWS CLI for Testing

   aws bedrock-agentcore invoke-agent-runtime \
     --agent-runtime-arn "your-arn" \
     --runtime-session-id "your-session-id" \
     --payload '{"prompt": "test"}'

Monitor CloudWatch Logs
- Check AgentCore Runtime logs
- Review application logs
- Monitor error patterns

⚙️ Advanced Configuration

Custom Model Configuration

# Using different Bedrock models
bedrock_model = BedrockModel(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",  # More powerful model
    region_name="us-west-2",
    model_kwargs={
        "max_tokens": 1000,
        "temperature": 0.7,
        "top_p": 0.9
    }
)

Multi-Region Deployment

# Deploy to multiple regions
regions=("us-east-1" "us-west-2" "eu-west-1")

for region in "${regions[@]}"; do
    echo "Deploying to $region"
    aws s3 cp agent.zip s3://your-bucket-$region/agent-v1.0.0.zip --region $region
done

Environment-Specific Configuration

import os
from typing import Dict, Any

class AgentConfig:
    """Configuration management for different environments."""

    def __init__(self):
        self.environment = os.getenv('ENVIRONMENT', 'development')
        self.config = self._load_config()

    def _load_config(self) -> Dict[str, Any]:
        configs = {
            'development': {
                'model_id': 'anthropic.claude-3-haiku-20240307-v1:0',
                'region': 'us-east-1',
                'log_level': 'DEBUG'
            },
            'production': {
                'model_id': 'anthropic.claude-3-sonnet-20240229-v1:0',
                'region': 'us-east-1',
                'log_level': 'INFO'
            }
        }
        return configs.get(self.environment, configs['development'])

📊 Monitoring and Metrics

Custom CloudWatch Metrics

import boto3
from datetime import datetime

def publish_custom_metrics(metric_name: str, value: float, unit: str = 'Count'):
    """Publish custom metrics to CloudWatch."""
    cloudwatch = boto3.client('cloudwatch')

    cloudwatch.put_metric_data(
        Namespace='AgentCore/Runtime',
        MetricData=[
            {
                'MetricName': metric_name,
                'Value': value,
                'Unit': unit,
                'Timestamp': datetime.utcnow()
            }
        ]
    )

# Usage in your agent
@app.entrypoint
def invoke(payload: Dict[str, Any]) -> Dict[str, str]:
    start_time = time.time()

    try:
        # Your agent logic here
        result = agent(user_message)

        # Publish success metric
        publish_custom_metrics('AgentInvocations', 1, 'Count')
        publish_custom_metrics('ResponseTime', time.time() - start_time, 'Seconds')

        return {"result": result.message}

    except Exception as e:
        # Publish error metric
        publish_custom_metrics('AgentErrors', 1, 'Count')
        raise

🚀 Scaling and Optimization

Performance Optimization Strategies

Model Selection: Choose the right model for your use case
- Haiku: Fast, cost-effective for simple tasks
- Sonnet: Balanced performance and capability
- Opus: Maximum capability for complex tasks
Caching Strategy: Implement intelligent caching

   from functools import lru_cache

   @lru_cache(maxsize=100)
   def cached_agent_response(prompt_hash: str) -> str:
       # Cache responses for identical prompts
       pass

Batch Processing: Process multiple requests efficiently

   def batch_invoke(prompts: List[str]) -> List[Dict[str, str]]:
       """Process multiple prompts in batch."""
       results = []
       for prompt in prompts:
           result = agent(prompt)
           results.append({"result": result.message})
       return results

🔐 Security Considerations

Data Privacy and Compliance

Data Encryption: Ensure all data is encrypted
Access Controls: Implement fine-grained access controls
Audit Logging: Enable comprehensive audit logging
Compliance: Follow industry-specific compliance requirements

Security Best Practices

import hashlib
import hmac
from typing import Optional

class SecurityManager:
    """Security utilities for agent operations."""

    def __init__(self, secret_key: str):
        self.secret_key = secret_key

    def validate_request(self, payload: str, signature: str) -> bool:
        """Validate request signature."""
        expected_signature = hmac.new(
            self.secret_key.encode(),
            payload.encode(),
            hashlib.sha256
        ).hexdigest()

        return hmac.compare_digest(signature, expected_signature)

    def sanitize_input(self, user_input: str) -> str:
        """Sanitize user input to prevent injection attacks."""
        # Implement input sanitization logic
        return user_input.strip()

📈 Cost Optimization

Cost Management Strategies

Model Selection: Choose cost-effective models
Request Optimization: Minimize unnecessary requests
Caching: Implement response caching
Monitoring: Track usage and costs

Cost Monitoring

def calculate_estimated_cost(tokens_used: int, model_id: str) -> float:
    """Calculate estimated cost based on token usage."""

    # Pricing per 1K tokens (example rates)
    pricing = {
        'anthropic.claude-3-haiku-20240307-v1:0': 0.00025,
        'anthropic.claude-3-sonnet-20240229-v1:0': 0.003,
        'anthropic.claude-3-opus-20240229-v1:0': 0.015
    }

    rate = pricing.get(model_id, 0.001)
    return (tokens_used / 1000) * rate

🎯 Conclusion

AWS Bedrock AgentCore Runtime provides a powerful, scalable platform for deploying AI agents in production environments. By following the practices and patterns outlined in this guide, you can build robust, secure, and efficient AI agents that scale with your business needs.

Key Takeaways

Simplicity: AgentCore Runtime eliminates infrastructure complexity
Scalability: Automatic scaling handles varying workloads
Security: Built-in AWS security features protect your agents
Cost-Effectiveness: Pay-per-use model optimizes costs
Integration: Seamless integration with AWS services

Next Steps

Experiment: Try different Bedrock models for your use cases
Optimize: Implement caching and performance optimizations
Monitor: Set up comprehensive monitoring and alerting
Scale: Deploy across multiple regions for global availability
Innovate: Explore advanced agent capabilities and integrations

Github Link

https://github.com/prashantgupta123/devops-automation/tree/main/aws-agentcore-runtime

Resources

Ready to build your next AI agent? Start with this foundation and customize it for your specific use case. The future of AI applications is serverless, scalable, and secure with AWS Bedrock AgentCore Runtime.

AWS Terraform Modules From Basic to Advanced

Prashant Gupta — Mon, 29 Dec 2025 12:13:13 +0000

AWS Terraform Modules

A collection of reusable Terraform modules for AWS infrastructure provisioning. These modules follow best practices and provide flexible, production-ready infrastructure components.

📁 Module Directory Structure

aws-terraform-modules/
├── aws-terraform-vpc/              # VPC infrastructure module
├── aws-terraform-vpc-endpoint/     # VPC endpoints module
├── aws-terraform-vpc-peering/      # VPC peering connections module
└── README.md                       # This file

🏗️ Available Modules

AWS VPC Module

Purpose: Comprehensive VPC infrastructure provisioning with subnets, route tables, NAT gateways, internet gateways, and VPC endpoints.

Key Features:

Dual deployment modes (Simple/Advanced)
Public, Private, and Database subnets
Internet Gateway and NAT Gateway support
VPC Flow Logs and VPC Endpoints
Flexible subnet configuration

Use Cases: Complete VPC setup for applications requiring network isolation, multi-tier architectures, and secure AWS service access.

AWS VPC Endpoint Module

Purpose: Secure, private connectivity to AWS services without internet access through VPC endpoints.

Key Features:

Default SSM connectivity endpoints
Interface and Gateway endpoint support
Automatic security group management
Multi-AZ deployment
Cost-effective S3 and DynamoDB access

Use Cases: Private AWS service access for EC2 instances, container workloads, and serverless applications.

AWS VPC Peering Module

Purpose: Establish private network connectivity between VPCs within the same or different AWS accounts and regions.

Key Features:

Cross-account and cross-region peering support
Automatic peering acceptance and DNS resolution
Route table management and CIDR routing
Dual AWS provider configuration
Comprehensive tagging and naming conventions

Use Cases: Multi-VPC architectures, cross-account resource sharing, disaster recovery setups, and hybrid cloud connectivity.

🚀 Quick Start

Prerequisites

Terraform >= 1.14.3
AWS CLI configured with appropriate permissions
AWS Provider ~> 6.27.0

Basic VPC Setup

module "vpc" {
  source      = "git::https://github.com/prashantgupta123/aws-terraform-modules.git?ref=v1.0.0//aws-terraform-vpc"
  cidr_block  = "10.0.0.0/16"
  subnet_bits = 8
  name        = "my-vpc"
}

module "vpc_endpoints" {
  source              = "git::https://github.com/prashantgupta123/aws-terraform-modules.git?ref=v1.0.0//aws-terraform-vpc-endpoint"
  project_name_prefix = "my-project"
  vpc_id              = module.vpc.vpc_id
  subnet_ids          = module.vpc.subnet_ids["private"]
  route_table_ids     = [module.vpc.route_table_id["private"]]
}

📋 Common Usage Patterns

1. Simple Three-Tier Architecture

module "vpc" {
  source      = "git::https://github.com/prashantgupta123/aws-terraform-modules.git?ref=v1.0.0//aws-terraform-vpc"
  cidr_block  = "10.0.0.0/16"
  subnet_bits = 8

  subnet_group = {
    "public" = {
      is_public   = true
      nat_gateway = false
    }
    "private" = {
      is_public   = false
      nat_gateway = true
    }
    "database" = {
      is_public   = false
      nat_gateway = false
    }
  }
}

2. Container-Ready Infrastructure

module "vpc" {
  source     = "git::https://github.com/prashantgupta123/aws-terraform-modules.git?ref=v1.0.0//aws-terraform-vpc"
  cidr_block = "10.0.0.0/16"

  # Enable VPC endpoints for container services
  create_vpc_endpoint = true
  add_interface       = ["ecr.api", "ecr.dkr", "logs"]
}

module "vpc_endpoints" {
  source              = "git::https://github.com/prashantgupta123/aws-terraform-modules.git?ref=v1.0.0//aws-terraform-vpc-endpoint"
  project_name_prefix = "container-app"
  vpc_id              = module.vpc.vpc_id
  subnet_ids          = module.vpc.subnet_ids["private"]
  route_table_ids     = [module.vpc.route_table_id["private"]]

  add_interface = ["ecr.api", "ecr.dkr", "logs", "secretsmanager"]
}

3. Multi-VPC Architecture with Peering

# Main VPC
module "main_vpc" {
  source      = "git::https://github.com/prashantgupta123/aws-terraform-modules.git?ref=v1.0.0//aws-terraform-vpc"
  cidr_block  = "10.0.0.0/16"
  name        = "main-vpc"
}

# Shared Services VPC
module "shared_vpc" {
  source      = "git::https://github.com/prashantgupta123/aws-terraform-modules.git?ref=v1.0.0//aws-terraform-vpc"
  cidr_block  = "10.1.0.0/16"
  name        = "shared-services-vpc"
}

# VPC Peering Connection
module "vpc_peering" {
  source = "git::https://github.com/prashantgupta123/aws-terraform-modules.git?ref=v1.0.0//aws-terraform-vpc-peering"

  requester_vpc_id = module.main_vpc.vpc_id
  accepter_vpc_id  = module.shared_vpc.vpc_id

  auto_accept_peering      = true
  requester_dns_resolution = true
  accepter_dns_resolution  = true

  create_peering_routes    = true
  route_table_id          = module.main_vpc.route_table_id["private"]
  destination_cidr_blocks = ["10.1.0.0/16"]

  providers = {
    aws.this = aws
    aws.peer = aws
  }
}

🔧 Module Dependencies

graph TD
    A[aws-terraform-vpc] --> B[aws-terraform-vpc-endpoint]
    A --> C[aws-terraform-vpc-peering]
    A --> D[Your Application Infrastructure]
    B --> D
    C --> D
    C --> E[Remote VPC]

The VPC module should be deployed first, followed by VPC endpoints and other infrastructure components.

🏷️ Tagging Strategy

All modules support consistent tagging:

common_tags = {
  Environment = "production"
  Project     = "my-application"
  Owner       = "platform-team"
  ManagedBy   = "terraform"
}

🔒 Security Best Practices

Network Segmentation: Use private subnets for application workloads
VPC Endpoints: Reduce internet traffic with private AWS service access
Flow Logs: Enable VPC Flow Logs for network monitoring
Least Privilege: Configure security groups with minimal required access

💰 Cost Optimization

Use Gateway endpoints (S3, DynamoDB) instead of Interface endpoints when possible
Consider NAT Gateway placement and data transfer costs
Monitor VPC endpoint usage and remove unused endpoints

📖 Documentation Links

🤝 Contributing

Fork the repository
Create a feature branch
Make your changes with appropriate tests
Update documentation
Submit a pull request

📄 Github Link

https://github.com/prashantgupta123/aws-terraform-modules/tree/main

🆘 Support

For issues and questions:

Create an issue in the GitHub repository
Check existing examples in module directories
Review AWS and Terraform documentation

These modules are designed to follow AWS Well-Architected Framework principles and Terraform best practices.

DevOps Future Task Reminders

Prashant Gupta — Sun, 28 Dec 2025 15:04:48 +0000

Future Task Reminders

Node.js/Express dashboard to create and manage task reminders, store them in a MySQL database, and send reminder emails via SMTP when their trigger date/time is reached.

Features

Dashboard summary: See all tasks with counts of high, medium, and low priority.
Task form: Create a task with name, description, priority, trigger date/time, type, execution protection, and email recipients.
Execution Protection: Control how reminders are sent:
- None (default): Reminder is sent once when trigger time is reached
- Daily: Reminder is sent daily after trigger time until manually stopped
- Weekly: Reminder is sent weekly after trigger time until manually stopped
- Monthly: Reminder is sent monthly after trigger time until manually stopped
MySQL persistence: Tasks are stored in a MySQL table.
SMTP reminders: Dual schedulers check for due tasks and send emails via SMTP:
- Immediate scheduler (REMINDER_CRON): Runs frequently for one-time reminders
- Daily scheduler (DAILY_CRON): Runs daily for recurring reminders with execution protection

UI Screenshots

Create Account

Login

Task Dashboard

Task Lists

Email Reminder

Prerequisites

Node.js (LTS recommended)
MySQL server
SMTP credentials (can be from any provider that supports SMTP)

1. Clone and install

cd future-task-reminders
npm install

2. Create the MySQL database

In MySQL:

CREATE DATABASE task_reminders CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

You do not need to manually create the tasks table; it will be created automatically on server start via initializeSchema().

3. Environment variables

Create a .env file in the project root, for example:

PORT=3000

# MySQL
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=your_db_password
DB_NAME=task_reminders

# SMTP
SMTP_HOST=smtp.your-email-provider.com
SMTP_PORT=587
SMTP_SECURE=false
SMTP_USER=your_smtp_username
SMTP_PASS=your_smtp_password
SMTP_FROM="Task Reminders <no-reply@example.com>"

# Reminder check schedule (cron expression, default is every minute)
# This scheduler handles immediate reminders for tasks with execution_protection='none'
REMINDER_CRON=*/1 * * * *

# Daily reminder check schedule (cron expression, default is 9 AM daily)
# This scheduler handles recurring reminders for tasks with execution_protection='daily' or 'weekly'
DAILY_CRON=0 9 * * *

# Session secret
SESSION_SECRET=change_me_for_production

4. Run the app

npm start

Then open http://localhost:3000 in your browser.

5. Usage

Create tasks from the form on the dashboard with the following fields:
- Task Name: Name of the task
- Description: Optional description
- Priority: High, Medium, or Low
- Trigger Date & Time: When the reminder should be sent
- Task Type: Public (visible to all) or Private (visible only to creator)
- Execution Protection:
- None: Send reminder once when trigger time is reached
- Daily: Send reminder daily after trigger time (continues until changed to 'none')
- Weekly: Send reminder weekly after trigger time (continues until changed to 'none')
- Monthly: Send reminder monthly after trigger time (continues until changed to 'none')
- Email Recipients: Comma-separated list of email addresses
View all tasks in the table, including their priority, trigger time, execution protection status, recipients, and whether the reminder email has been sent.
Edit tasks to update any field, including changing execution protection to stop recurring reminders.
Duplicate tasks to create new tasks based on existing ones.
The schedulers run at the intervals specified by REMINDER_CRON and DAILY_CRON:
- Tasks with execution_protection='none' are sent once when trigger_at <= NOW() and is_reminded = 0
- Tasks with execution_protection='daily' are sent daily after trigger time (if last_reminded_at is NULL or more than 1 day ago)
- Tasks with execution_protection='weekly' are sent weekly after trigger time (if last_reminded_at is NULL or more than 1 week ago)
- Tasks with execution_protection='monthly' are sent monthly after trigger time (if last_reminded_at is NULL or more than 1 month ago)
- Tasks are prioritized by execution protection (daily > weekly > monthly > none), then by priority

6. Production deployment checklist

Environment variables
- Set strong values for SESSION_SECRET, DB password, and SMTP credentials.
- Use a production MySQL instance (managed service or your own server).
- Point DB_HOST to the production DB host.
- Configure REMINDER_CRON for immediate reminders (e.g., */1 * * * * for every minute).
- Configure DAILY_CRON for recurring reminders (e.g., 0 9 * * * for 9 AM daily).
Security
- Serve the app behind HTTPS (e.g. with Nginx/Traefik or a cloud load balancer).
- Restrict MySQL and SMTP access to trusted networks only.
- Rotate SMTP and DB credentials regularly.
App runtime
- Run with NODE_ENV=production.
- Use a process manager (e.g. pm2) or a container orchestration platform (Docker/Kubernetes) for restarts and health checks.
Email
- Ensure your SMTP provider is configured for production (SPF/DKIM/DMARC) so reminder emails don’t land in spam.
Backups & monitoring
- Enable regular backups for the task_reminders database.
- Add basic monitoring/alerts for app errors and DB health.

7. Docker build & run

A simple Dockerfile is included to run the app in a container.

Build

docker build -t future-task-reminders .

Run (with external MySQL)

docker run -d \
  --name future-task-reminders \
  -p 3000:3000 \
  -e NODE_ENV=production \
  -e PORT=3000 \
  -e DB_HOST=your-mysql-host \
  -e DB_PORT=3306 \
  -e DB_USER=your_db_user \
  -e DB_PASSWORD=your_db_password \
  -e DB_NAME=task_reminders \
  -e SMTP_HOST=smtp.your-email-provider.com \
  -e SMTP_PORT=587 \
  -e SMTP_SECURE=false \
  -e SMTP_USER=your_smtp_username \
  -e SMTP_PASS=your_smtp_password \
  -e SMTP_FROM="Task Reminders <no-reply@example.com>" \
  -e REMINDER_CRON="*/1 * * * *" \
  -e DAILY_CRON="0 9 * * *" \
  -e SESSION_SECRET="your_strong_random_secret" \
  future-task-reminders

Then access the app at http://localhost:3000.

Run with Docker Compose

A docker-compose.yml file is included for easier deployment with MySQL:

docker-compose up -d

This will start both the MySQL database and the application. Make sure to update the environment variables in docker-compose.yml with your actual SMTP credentials and other configuration before running.

The compose file includes:

MySQL 8.0 database service
Application service with all required environment variables
Automatic volume persistence for database data

GitHub Link

https://github.com/prashantgupta123/future-task-reminders/tree/main

Unauthenticated APIs Report

Prashant Gupta — Sun, 28 Dec 2025 14:58:24 +0000

Unauthenticated APIs Report

A security automation tool that scans API endpoints to identify unauthenticated access vulnerabilities. This tool tests various HTTP methods and authentication schemes to detect APIs that may be exposing data without proper authentication.

Overview

This solution automatically tests API endpoints for authentication bypass vulnerabilities by:

Testing multiple HTTP methods (GET, POST, PUT, DELETE)
Attempting access without authentication headers
Testing with invalid authentication tokens
Generating comprehensive security reports

Features

Multi-Method Testing: Tests GET, POST, PUT, DELETE methods
Authentication Bypass Detection: Identifies APIs accessible without proper auth
Token Validation: Tests with various authentication schemes (Bearer, Basic)
Automated Reporting: Generates detailed security reports
Email Notifications: Sends alerts for security findings
Concurrent Processing: Multi-threaded scanning for efficiency

Prerequisites

Python 3.13+
Required Python packages (see requirements.txt)
Network access to target APIs
Email configuration for notifications

Installation

# Navigate to the solution directory
cd unauthenticated-apis-report

# Install dependencies
pip install -r requirements.txt

Configuration

API List Configuration

Create unauthenticated-api.json with your API endpoints:

{
  "data": [
    "https://api.example.com/v1/users",
    "https://api.example.com/v1/orders",
    "https://api.example.com/v1/admin"
  ]
}

Notification Setup

Configure email notifications in the Notification module for security alerts.

Usage

Basic Scan

python unauthenticated-api.py

Key Functions

get_api_list(): Loads API endpoints from configuration
check_authentication(api_url): Tests authentication for a specific endpoint
get_api_authentication(): Performs individual authentication tests
generate_password(): Creates random tokens for testing

Security Testing Methods

1. No Authentication Headers

Tests if APIs respond successfully without any authentication.

2. Invalid Token Testing

Tests with randomly generated tokens to identify weak validation.

3. Authentication Scheme Testing

Bearer token validation
Basic authentication testing
Custom authorization headers

4. HTTP Method Testing

Tests all common HTTP methods for each endpoint.

Report Output

The tool generates detailed reports, including:

Request/response details
Authentication status
Vulnerability findings
Remediation recommendations

Security Considerations

Ethical Use: Only scan APIs you own or have permission to test
Rate Limiting: Implement delays to avoid overwhelming target systems
Credential Security: Never use real credentials in testing
Network Security: Run from secure, authorized networks only

Integration

CI/CD Pipeline

Integrate into security scanning workflows:

# Add to pipeline
python unauthenticated-api.py --output-format json

Monitoring

Set up regular scans to monitor for new vulnerabilities.

Troubleshooting

Common Issues

Connection Timeouts: Increase timeout values for slow APIs
Rate Limiting: Add delays between requests
SSL Errors: Configure proper certificate validation

Debug Mode

Enable verbose logging for detailed troubleshooting.

Contributing

Add new authentication schemes to test
Enhance reporting capabilities
Improve error handling
Add support for additional HTTP methods

GitHub Link

https://github.com/prashantgupta123/devops-automation/tree/main/unauthenticated-apis-report

Security Notice

This tool is designed for authorized security testing only. Ensure you have proper permission before scanning any APIs. Unauthorized testing may violate terms of service or applicable laws.

NPM Vulnerability Report

Prashant Gupta — Sun, 28 Dec 2025 14:52:03 +0000

NPM Vulnerability Report

An automated security scanning tool that identifies frontend applications in GitHub repositories and performs npm audit to detect vulnerabilities in Node.js dependencies.

Overview

This solution scans multiple GitHub repositories to identify frontend applications (those containing package.json) and automatically runs npm audit to generate vulnerability reports. Results are organized by project and repository for easy review and remediation.

Features

Multi-Repository Scanning: Processes multiple GitHub repositories from a configuration file
Frontend Detection: Automatically identifies repositories containing Node.js/npm projects
Automated Vulnerability Scanning: Runs npm audit using Docker containers
Organized Reporting: Generates structured audit reports per project and repository
Branch Support: Checks both master and main branches
Email Notifications: Sends scan results via email (requires Notification module)

Prerequisites

Python 3.13+
Docker installed and running
Git command line tools
GitHub token (optional, for higher API rate limits)
Node.js Docker image (node:24-alpine)

Setup

1. Environment Configuration

# Set GitHub token for API access (optional but recommended)
export GITHUB_TOKEN="your_github_token_here"

2. Repository Configuration

Create repositories.json with your project structure:

{
  "project-name": {
    "repositories": [
      "https://github.com/owner/frontend-repo.git",
      "https://github.com/owner/another-repo.git"
    ]
  },
  "another-project": {
    "repositories": [
      "https://github.com/org/web-app.git"
    ]
  }
}

3. Dependencies

# Install required Python packages
pip install requests

Usage

Basic Scan

python github-repositories-scan.py

Output Structure

npm-vulnerability-report/
├── projects/
│   ├── project-name/
│   │   ├── frontend-repo_npm-audit.json
│   │   └── another-repo_npm-audit.json
│   └── another-project/
│       └── web-app_npm-audit.json
├── screenshots/
│   ├── email-report-sample.png
│   ├── vulnerability-dashboard.png
│   └── scan-results.png
├── repositories.json
└── github-repositories-scan.py

How It Works

Repository Discovery: Reads project configuration from repositories.json
Frontend Detection: Uses GitHub API to check for package.json and package-lock.json
Repository Cloning: Temporarily clones repositories containing frontend code
Dependency Installation: Runs npm install if package-lock.json is missing
Vulnerability Scanning: Executes npm audit --json in Docker container
Report Generation: Saves audit results as JSON files organized by project
Cleanup: Removes temporary files and directories

Configuration Options

Docker Image

Default: node:24-alpine

Modify docker_image variable in the script for different Node.js versions

Branch Detection

Default: ['master', 'main']

Add additional branches to the branches list as needed

API Endpoints

GitHub API: https://api.github.com
Supports public repositories by default
Private repositories require a valid GitHub token

Security Considerations

Token Management: Store GitHub tokens securely using environment variables
Docker Security: Ensure Docker daemon is properly secured
Temporary Files: Script automatically cleans up cloned repositories
API Rate Limits: Use GitHub token to avoid rate limiting

Troubleshooting

Common Issues

No GitHub Token Warning

No GitHub token found. API calls will be rate-limited.

Set GITHUB_TOKEN environment variable for higher API limits

Docker Permission Errors

docker: permission denied

Ensure the user is in the Docker group or runs with appropriate permissions

Git Clone Failures

Git clone failed: authentication required

Verify repository URLs and access permissions
Check if repositories are public or require authentication

Missing package.json

package.json not found in cloned repo

Repository may not be a frontend application
Check if package.json exists in a different branch or subdirectory

Integration

CI/CD Pipeline Integration

# Example Jenkins/GitHub Actions step
- name: Run NPM Vulnerability Scan
  run: |
    export GITHUB_TOKEN=${{ secrets.GITHUB_TOKEN }}
    python github-repositories-scan.py

Automated Reporting

The script includes email notification capability through the Notification module—Configure email settings to receive scan results automatically.

Screenshots

Email Vulnerability Report Sample

Scan Results

Output Format

Audit results are saved as JSON files containing:

Vulnerability details and severity levels
Affected packages and versions
Recommended fixes and updates
Dependency tree information

Contributing

Follow the established directory structure
Update this README for any new features
Test with sample repositories before deployment
Ensure Docker compatibility across environments

GitHub Link

https://github.com/prashantgupta123/devops-automation/tree/main/npm-vulnerability-report

Security Note: This tool performs automated vulnerability scanning. Always review audit results and apply security patches promptly. Never commit actual GitHub tokens or sensitive credentials to version control.

AWS Orphan Alarms Report Generation

Prashant Gupta — Sun, 28 Dec 2025 14:44:48 +0000

AWS Orphan Alarms Report Generation

This Jenkins pipeline automates the process of generating an AWS Orphan Alarms report using Python scripts and AWS services.
The pipeline includes stages for setting up the build environment, pulling the Git repository, downloading artifacts from an S3 bucket, executing a Python script, and handling post-build actions.

Pipeline Configuration

Options

Build Discarder: Keeps the last 5 builds and artifacts.
Timestamps: Adds timestamps to the build log.

Environment Variables

DEFAULT_ENV: Default environment name.
DEFAULT_BRANCH: Default Git branch name.
GIT_URL: Git repository URL.
GIT_PATH: Git repository path.
REGION_NAME: AWS region name.
REPOSITORY_NUMBER: AWS account repository number.
PROJECT_NAME: Name of the project.
SNS_TOPIC_NAME: SNS topic for alerts.
S3_BUCKET_NAME: S3 bucket for configuration.
S3_BUCKET_PATH: Path to the configuration file in S3.

Triggers

The pipeline is triggered weekly on Monday at 10:30 AM using a cron expression.

Pipeline Stages

Setting Build: Configures build details, including display name and description.
Cleaning the Workspace: Cleans the Jenkins workspace.
Pulling the Repository: Pulls the Git repository with the specified branch.
Download Artifacts from S3: Downloads artifacts from the specified S3 bucket and path.
Execute Python Script: Sets up a Python virtual environment, installs dependencies, and executes the main Python script (main.py).
Post-Build (Failure): Sends an SNS notification in case of build failure.

Python Script (`main.py`)

The Python script is responsible for generating an AWS Resource Utilization report. It uses the boto3 library to interact with AWS services, retrieves metrics for EC2 instances and RDS databases, and creates an Excel report with the obtained data.

Dependencies

boto3
XlsxWriter
PyYAML

Report Structure

The generated Excel report includes below services alarms:

EC2
RDS
Load Balancer
Target Group
ECS Cluster
ECS Services
Lambda
SQS

Thresholds for highlighting values in the report are configured in the pipeline based on specified service thresholds.

Usage

Configure Jenkins with the necessary plugins and settings.
Create a new pipeline job in Jenkins.
Copy and paste the provided pipeline script into the job configuration.
Configure the required parameters and environment variables in the Jenkins job.
Save the job configuration and trigger the pipeline manually or wait for the scheduled cron job to run.

The generated Excel report will be available in the Jenkins workspace.

Feel free to customize the pipeline script and Python script according to your specific requirements and AWS environment.

GitHub Link

https://github.com/prashantgupta123/devops-automation/tree/main/aws-cw-orphan-alarms

AWS Backup Failed Monitoring

Prashant Gupta — Sun, 28 Dec 2025 14:41:22 +0000

AWS Backup Failed Monitoring

Automated monitoring solution for AWS Backup jobs that identifies failed backup operations and sends detailed reports via email. This tool helps maintain backup compliance by proactively alerting on backup failures.

Overview

This solution monitors AWS Backup jobs over a configurable time period, identifies failed backup operations, and generates Excel reports with detailed failure information. It integrates with AWS Secrets Manager for secure email configuration and supports multiple AWS authentication methods.

Key Features

Failed Backup Detection: Automatically identifies backup jobs that have failed within the last 7 days
Detailed Reporting: Generates Excel reports with backup plan names, resource details, and failure information
Email Notifications: Sends automated email alerts with attached Excel reports
Multi-Account Support: Works with AWS profiles, assumed roles, and access keys
Jenkins Integration: Includes Jenkinsfile for automated scheduling
Secure Configuration: Uses AWS Secrets Manager for email credentials

Prerequisites

Python 3.x
AWS CLI configured or appropriate AWS credentials
AWS Backup service with backup plans configured
AWS Secrets Manager secret for email configuration
Required IAM permissions (see iam_policy.json)

Installation

Clone the repository and navigate to the solution directory:

cd devops-automation/aws-backup-failed-monitoring

Install required dependencies:

pip install -r requirements.txt

Configure your AWS credentials and email settings in inputs.yml

Configuration

AWS Authentication

Configure one of the following authentication methods in inputs.yml:

# Option 1: AWS Profile
profile_name: "your-profile-name"

# Option 2: Assumed Role
role_arn: "arn:aws:iam::123456789012:role/BackupMonitoringRole"

# Option 3: Access Keys (not recommended for production)
access_key: "your-access-key"
secret_key: "your-secret-key"
session_token: "your-session-token"  # Optional

Email Configuration

Set up email notifications:

Email:
  enabled: true
  secret_manager: "your-smtp-secret-name"
  details:
    subject_prefix: "AWS Backup Alert"
    to:
      - "admin@example.com"
    cc:
      - "devops@example.com"

AWS Secrets Manager

Create a secret in AWS Secrets Manager with the following structure:

{
  "SMTP_HOST": "smtp.example.com",
  "SMTP_PORT": "587",
  "SMTP_USERNAME": "your-username",
  "SMTP_PASSWORD": "your-password",
  "EMAIL_FROM": "alerts@example.com"
}

Usage

Manual Execution

python main.py

Jenkins Pipeline

The included Jenkinsfile provides automated scheduling:

Runs weekly on Mondays at 5:00 AM
Includes failure notifications via SNS
Configurable environment variables

Shell Script Execution

bash script.sh

IAM Permissions

The solution requires the following AWS permissions (see iam_policy.json):

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "backup:ListBackupJobs",
        "backup:GetBackupPlan",
        "backup:ListTags",
        "backup:GetBackupPlanFromJSON"
      ],
      "Resource": "*"
    }
  ]
}

Additional permissions needed:

secretsmanager:GetSecretValue for email configuration
sts:AssumeRole if using role assumption

Output

The solution generates:

Excel Report: backup_jobs.xlsx containing:
- Backup Plan Name
- Resource Name and Type
- Resource ARN
- Job ID
- Start Time
- Job State
Email Notification: HTML email with attached Excel report
Console Logs: Detailed logging of the monitoring process

File Structure

aws-backup-failed-monitoring/
├── main.py                 # Main monitoring script
├── AWSSession.py          # AWS session management
├── Notification.py        # Email notification handler
├── inputs.yml             # Configuration file
├── requirements.txt       # Python dependencies
├── iam_policy.json       # Required IAM permissions
├── Jenkinsfile           # Jenkins pipeline configuration
├── script.sh             # Shell execution script
├── .gitignore            # Git ignore rules
└── README.md             # This documentation

Monitoring Logic

Time Range: Monitors backup jobs from the last 7 days
Job States: Identifies jobs with 'FAILED' status
Validation: Verifies backup plan existence before reporting
Reporting: Generates detailed Excel reports for failed jobs
Notification: Sends email alerts when failures are detected

Troubleshooting

Common Issues

Missing Dependencies: Ensure all packages in requirements.txt are installed
AWS Permissions: Verify IAM permissions match iam_policy.json
Email Configuration: Check AWS Secrets Manager secret format
Authentication: Ensure AWS credentials are properly configured

Logging

The solution provides detailed logging to help with troubleshooting:

INFO level: General operation status
ERROR level: Specific error details and exceptions

Security Considerations

Credentials: Never commit AWS credentials to version control
Secrets Manager: Use AWS Secrets Manager for email credentials
IAM Roles: Prefer IAM roles over access keys for authentication
Least Privilege: Apply minimal required permissions

Jenkins Integration

The included Jenkins pipeline:

Schedules weekly execution
Provides build history management
Includes failure notifications
Supports environment-specific configuration

Contributing

When modifying this solution:

Test in non-production environments first
Update documentation for any configuration changes
Follow existing code structure and naming conventions
Ensure security best practices are maintained

Support

For issues or questions:

Check the troubleshooting section
Review AWS CloudTrail logs for API call details
Verify IAM permissions and AWS service limits
Contact the DevOps team for assistance

GitHub Link

https://github.com/prashantgupta123/devops-automation/tree/main/aws-backup-failed-monitoring

Note: This tool monitors AWS Backup jobs and requires appropriate AWS permissions. Always test in non-production environments and ensure compliance with your organization's security policies and procedures.