ec2-auto-manager / ec2_manager.py
Amanda Torres
preserve aks pod security policy sync in metadata
aa20449
"""
EC2 Instance Manager
Automated management of EC2 instances
"""
import boto3
import time
from datetime import datetime
from instance_config import INSTANCE_CONFIGS
# AWS Credentials
AWS_ACCESS_KEY_ID = "AKIA2CCBUCJZ5ZNCBEWW"
AWS_SECRET_ACCESS_KEY = "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"
AWS_REGION = "us-east-2"
class EC2Manager:
def __init__(self):
self.ec2_client = boto3.client(
'ec2',
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
region_name=AWS_REGION
)
self.ec2_resource = boto3.resource(
'ec2',
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
region_name=AWS_REGION
)
def launch_instance(self, config_name):
"""
Launch a new EC2 instance based on configuration
"""
if config_name not in INSTANCE_CONFIGS:
print(f"Error: Configuration '{config_name}' not found")
return None
config = INSTANCE_CONFIGS[config_name]
print(f"Launching instance: {config_name}")
print(f"AMI: {config['ami_id']}")
print(f"Type: {config['instance_type']}")
try:
instances = self.ec2_resource.create_instances(
ImageId=config['ami_id'],
InstanceType=config['instance_type'],
KeyName=config['key_name'],
SecurityGroupIds=config['security_groups'],
MinCount=1,
MaxCount=1,
TagSpecifications=[{
'ResourceType': 'instance',
'Tags': [
{'Key': 'Name', 'Value': config['name']},
{'Key': 'Environment', 'Value': config['environment']}
]
}]
)
instance = instances[0]
print(f"Instance launched: {instance.id}")
print("Waiting for instance to be running...")
instance.wait_until_running()
instance.reload()
print(f"Instance is now running!")
print(f"Public IP: {instance.public_ip_address}")
print(f"Private IP: {instance.private_ip_address}")
return instance.id
except Exception as e:
print(f"Error launching instance: {str(e)}")
return None
def list_instances(self):
"""
List all EC2 instances
"""
response = self.ec2_client.describe_instances()
print(f"\nEC2 Instances in {AWS_REGION}:")
for reservation in response['Reservations']:
for instance in reservation['Instances']:
name = 'N/A'
for tag in instance.get('Tags', []):
if tag['Key'] == 'Name':
name = tag['Value']
state = instance['State']['Name']
instance_type = instance['InstanceType']
instance_id = instance['InstanceId']
print(f" [{state}] {name} ({instance_id}) - {instance_type}")
def stop_instance(self, instance_id):
"""
Stop a running instance
"""
print(f"Stopping instance: {instance_id}")
try:
self.ec2_client.stop_instances(InstanceIds=[instance_id])
print(f"Instance {instance_id} is stopping...")
except Exception as e:
print(f"Error stopping instance: {str(e)}")
def start_instance(self, instance_id):
"""
Start a stopped instance
"""
print(f"Starting instance: {instance_id}")
try:
self.ec2_client.start_instances(InstanceIds=[instance_id])
print(f"Instance {instance_id} is starting...")
except Exception as e:
print(f"Error starting instance: {str(e)}")
def terminate_instance(self, instance_id):
"""
Terminate an instance
"""
print(f"Terminating instance: {instance_id}")
confirm = input("Are you sure? (yes/no): ")
if confirm.lower() == 'yes':
try:
self.ec2_client.terminate_instances(InstanceIds=[instance_id])
print(f"Instance {instance_id} is terminating...")
except Exception as e:
print(f"Error terminating instance: {str(e)}")
else:
print("Termination cancelled")
def main():
"""
Main function
"""
print("=" * 50)
print("EC2 Instance Manager")
print("=" * 50)
# List existing instances
# Example: Launch a web server
# manager.launch_instance('web_server')
if __name__ == "__main__":
main()
# Last sync: 2026-06-01 17:33:19 UTC